<!DOCTYPE html><html lang="zh-CN" data-theme="light"><head><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no"><title>数据采集系统 Flume | Oct25-X</title><meta name="keywords" content="Flume"><meta name="author" content="Oct25-X"><meta name="copyright" content="Oct25-X"><meta name="format-detection" content="telephone=no"><meta name="theme-color" content="#ffffff"><meta name="description" content="Flume 概述一、Flume 定义Flume 是 Cloudera 提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统。Flume基于流式架构，灵活简单。 为什么选择使用 Flume？ Flume 最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到 HDFS，或者是 Kafka等。 二、Flume 基础架构 AgentAgent 是一个 JVM 进程，它以事件的形式将">
<meta property="og:type" content="article">
<meta property="og:title" content="数据采集系统 Flume">
<meta property="og:url" content="https://oct25-x.atomgit.net/oct25-xxxxx/2024/01/03/42%E6%95%B0%E6%8D%AE%E9%87%87%E9%9B%86%E7%B3%BB%E7%BB%9F-Flume/index.html">
<meta property="og:site_name" content="Oct25-X">
<meta property="og:description" content="Flume 概述一、Flume 定义Flume 是 Cloudera 提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统。Flume基于流式架构，灵活简单。 为什么选择使用 Flume？ Flume 最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到 HDFS，或者是 Kafka等。 二、Flume 基础架构 AgentAgent 是一个 JVM 进程，它以事件的形式将">
<meta property="og:locale" content="zh_CN">
<meta property="og:image" content="https://oct25-x.atomgit.net/oct25-xxxxx/img/cover_img/ctt42.png">
<meta property="article:published_time" content="2024-01-03T00:30:37.000Z">
<meta property="article:modified_time" content="2024-02-17T14:24:42.827Z">
<meta property="article:author" content="Oct25-X">
<meta property="article:tag" content="Flume">
<meta name="twitter:card" content="summary">
<meta name="twitter:image" content="https://oct25-x.atomgit.net/oct25-xxxxx/img/cover_img/ctt42.png"><link rel="shortcut icon" href="/oct25-xxxxx/img/logo.jpg"><link rel="canonical" href="https://oct25-x.atomgit.net/oct25-xxxxx/2024/01/03/42%E6%95%B0%E6%8D%AE%E9%87%87%E9%9B%86%E7%B3%BB%E7%BB%9F-Flume/"><link rel="preconnect" href="//cdn.jsdelivr.net"/><link rel="preconnect" href="//busuanzi.ibruce.info"/><link rel="stylesheet" href="/oct25-xxxxx/css/index.css"><link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@fortawesome/fontawesome-free@6/css/all.min.css" media="print" onload="this.media='all'"><link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@fancyapps/ui/dist/fancybox.css" media="print" onload="this.media='all'"><script>const GLOBAL_CONFIG = { 
  root: '/oct25-xxxxx/',
  algolia: undefined,
  localSearch: {"path":"search.json","languages":{"hits_empty":"找不到您查询的内容：${query}"}},
  translate: {"defaultEncoding":2,"translateDelay":0,"msgToTraditionalChinese":"繁","msgToSimplifiedChinese":"簡"},
  noticeOutdate: undefined,
  highlight: {"plugin":"highlighjs","highlightCopy":true,"highlightLang":true,"highlightHeightLimit":false},
  copy: {
    success: '复制成功',
    error: '复制错误',
    noSupport: '浏览器不支持'
  },
  relativeDate: {
    homepage: false,
    post: false
  },
  runtime: '',
  date_suffix: {
    just: '刚刚',
    min: '分钟前',
    hour: '小时前',
    day: '天前',
    month: '个月前'
  },
  copyright: undefined,
  lightbox: 'fancybox',
  Snackbar: undefined,
  source: {
    justifiedGallery: {
      js: 'https://cdn.jsdelivr.net/npm/flickr-justified-gallery@2/dist/fjGallery.min.js',
      css: 'https://cdn.jsdelivr.net/npm/flickr-justified-gallery@2/dist/fjGallery.min.css'
    }
  },
  isPhotoFigcaption: false,
  islazyload: false,
  isAnchor: false
}</script><script id="config-diff">var GLOBAL_CONFIG_SITE = {
  title: '数据采集系统 Flume',
  isPost: true,
  isHome: false,
  isHighlightShrink: false,
  isToc: true,
  postUpdate: '2024-02-17 22:24:42'
}</script><noscript><style type="text/css">
  #nav {
    opacity: 1
  }
  .justified-gallery img {
    opacity: 1
  }

  #recent-posts time,
  #post-meta time {
    display: inline !important
  }
</style></noscript><script>(win=>{
    win.saveToLocal = {
      set: function setWithExpiry(key, value, ttl) {
        if (ttl === 0) return
        const now = new Date()
        const expiryDay = ttl * 86400000
        const item = {
          value: value,
          expiry: now.getTime() + expiryDay,
        }
        localStorage.setItem(key, JSON.stringify(item))
      },

      get: function getWithExpiry(key) {
        const itemStr = localStorage.getItem(key)

        if (!itemStr) {
          return undefined
        }
        const item = JSON.parse(itemStr)
        const now = new Date()

        if (now.getTime() > item.expiry) {
          localStorage.removeItem(key)
          return undefined
        }
        return item.value
      }
    }
  
    win.getScript = url => new Promise((resolve, reject) => {
      const script = document.createElement('script')
      script.src = url
      script.async = true
      script.onerror = reject
      script.onload = script.onreadystatechange = function() {
        const loadState = this.readyState
        if (loadState && loadState !== 'loaded' && loadState !== 'complete') return
        script.onload = script.onreadystatechange = null
        resolve()
      }
      document.head.appendChild(script)
    })
  
      win.activateDarkMode = function () {
        document.documentElement.setAttribute('data-theme', 'dark')
        if (document.querySelector('meta[name="theme-color"]') !== null) {
          document.querySelector('meta[name="theme-color"]').setAttribute('content', '#0d0d0d')
        }
      }
      win.activateLightMode = function () {
        document.documentElement.setAttribute('data-theme', 'light')
        if (document.querySelector('meta[name="theme-color"]') !== null) {
          document.querySelector('meta[name="theme-color"]').setAttribute('content', '#ffffff')
        }
      }
      const t = saveToLocal.get('theme')
    
          if (t === 'dark') activateDarkMode()
          else if (t === 'light') activateLightMode()
        
      const asideStatus = saveToLocal.get('aside-status')
      if (asideStatus !== undefined) {
        if (asideStatus === 'hide') {
          document.documentElement.classList.add('hide-aside')
        } else {
          document.documentElement.classList.remove('hide-aside')
        }
      }
    
    const detectApple = () => {
      if(/iPad|iPhone|iPod|Macintosh/.test(navigator.userAgent)){
        document.documentElement.classList.add('apple')
      }
    }
    detectApple()
    })(window)</script><link rel="stylesheet" href="/css/rightMenu.css"><meta name="generator" content="Hexo 6.1.0"></head><body><div id="sidebar"><div id="menu-mask"></div><div id="sidebar-menus"><div class="avatar-img is-center"><img src="/oct25-xxxxx/img/avatar.jpg" onerror="onerror=null;src='/img/friend_404.gif'" alt="avatar"/></div><div class="site-data is-center"><div class="data-item"><a href="/oct25-xxxxx/archives/"><div class="headline">文章</div><div class="length-num">52</div></a></div><div class="data-item"><a href="/oct25-xxxxx/tags/"><div class="headline">标签</div><div class="length-num">17</div></a></div><div class="data-item"><a href="/oct25-xxxxx/categories/"><div class="headline">分类</div><div class="length-num">8</div></a></div></div><hr/><div class="menus_items"><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/"><i class="fa-fw fas fa-home"></i><span> 首页</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/archives/"><i class="fa-fw fas fa-archive"></i><span> 归档</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/tags/"><i class="fa-fw fas fa-tags"></i><span> 标签</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/categories/"><i class="fa-fw fas fa-folder-open"></i><span> 分类</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/about/"><i class="fa-fw fas fa-heart"></i><span> 关于</span></a></div></div></div></div><div class="post" id="body-wrap"><header class="post-bg" id="page-header" style="background-image: url('/oct25-xxxxx/img/top_img/top_bg16.jpg')"><nav id="nav"><span id="blog_name"><a id="site-name" href="/oct25-xxxxx/">Oct25-X</a></span><div id="menus"><div id="search-button"><a class="site-page social-icon search"><i class="fas fa-search fa-fw"></i><span> 搜索</span></a></div><div class="menus_items"><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/"><i class="fa-fw fas fa-home"></i><span> 首页</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/archives/"><i class="fa-fw fas fa-archive"></i><span> 归档</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/tags/"><i class="fa-fw fas fa-tags"></i><span> 标签</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/categories/"><i class="fa-fw fas fa-folder-open"></i><span> 分类</span></a></div><div class="menus_item"><a class="site-page" href="/oct25-xxxxx/about/"><i class="fa-fw fas fa-heart"></i><span> 关于</span></a></div></div><div id="toggle-menu"><a class="site-page"><i class="fas fa-bars fa-fw"></i></a></div></div></nav><div id="post-info"><h1 class="post-title">数据采集系统 Flume</h1><div id="post-meta"><div class="meta-firstline"><span class="post-meta-date"><i class="far fa-calendar-alt fa-fw post-meta-icon"></i><span class="post-meta-label">发表于</span><time class="post-meta-date-created" datetime="2024-01-03T00:30:37.000Z" title="发表于 2024-01-03 08:30:37">2024-01-03</time><span class="post-meta-separator">|</span><i class="fas fa-history fa-fw post-meta-icon"></i><span class="post-meta-label">更新于</span><time class="post-meta-date-updated" datetime="2024-02-17T14:24:42.827Z" title="更新于 2024-02-17 22:24:42">2024-02-17</time></span><span class="post-meta-categories"><span class="post-meta-separator">|</span><i class="fas fa-inbox fa-fw post-meta-icon"></i><a class="post-meta-categories" href="/oct25-xxxxx/categories/%E5%A4%A7%E6%95%B0%E6%8D%AE%E7%BB%84%E4%BB%B6/">大数据组件</a></span></div><div class="meta-secondline"><span class="post-meta-separator">|</span><span class="post-meta-pv-cv" id="" data-flag-title="数据采集系统 Flume"><i class="far fa-eye fa-fw post-meta-icon"></i><span class="post-meta-label">阅读量:</span><span id="busuanzi_value_page_pv"></span></span></div></div></div></header><main class="layout" id="content-inner"><div id="post"><article class="post-content" id="article-container"><h1 id="Flume-概述"><a href="#Flume-概述" class="headerlink" title="Flume 概述"></a>Flume 概述</h1><h2 id="一、Flume-定义"><a href="#一、Flume-定义" class="headerlink" title="一、Flume 定义"></a>一、Flume 定义</h2><p>Flume 是 Cloudera 提供的一个高可用的，高可靠的，分布式的<code>海量日志采集、聚合和传输</code>的系统。Flume基于流式架构，灵活简单。</p>
<p><strong>为什么选择使用 Flume？</strong></p>
<p>Flume 最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到 HDFS，或者是 Kafka等。</p>
<h2 id="二、Flume-基础架构"><a href="#二、Flume-基础架构" class="headerlink" title="二、Flume 基础架构"></a>二、Flume 基础架构</h2><p><img src="/oct25-xxxxx/img/flume/flume01.png" alt="Flume 基础架构"></p>
<h3 id="Agent"><a href="#Agent" class="headerlink" title="Agent"></a>Agent</h3><p>Agent 是一个 JVM 进程，它以事件的形式将数据从源头送至目的。</p>
<p>Agent主要有3个部分组成，<code>Source、Channel、Sink</code>。</p>
<h3 id="Source"><a href="#Source" class="headerlink" title="Source"></a>Source</h3><p>Source 是负责接 收数据到 Flume Agent 的组件。Source 组件可以处理各种类型、各种格式的日志数据，包括 Avro Source、Thrift Source、Exec Source、JMS Source、Spooling Directory Source、Taildir Source、Kafka Source、NetCat TCP&#x2F;UDP Source、Sequence Generator Source、Syslog Source、Http Source 和 Legacy Source 等。</p>
<h3 id="Channel"><a href="#Channel" class="headerlink" title="Channel"></a>Channel</h3><p>Channel 是位于 Source 和 Sink 之间的缓冲区。因此，Channel 允许 Source 和 Sink 运作在不同的速率上。<code>Channel 是线程安全的</code>，可以同时处理几个 Source 的写入操作和几个 Sink 的读取操作。</p>
<p>Flume 有以下几种类型 Channel：Memory Channel、File Channel、Kafka Channel 和 JDBC Channel。</p>
<p><code>Memory Channel 是内存中的队列</code>。Memory Channel 在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么 Memory Channel 就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。</p>
<p><code>File Channel 将所有 Event 写到磁盘</code>。因此在程序关闭或机器宕机的情况下不会丢失数据。</p>
<p><code>Kafka Channel 将所有 Event 存储在 Kafka 集群中</code>。Kafka 集群需要单独部署，并且需要提供高可用性和副本机制，以便在 Kafka 挂掉时不会影响 Event 存储。</p>
<p><code>JDBC Channel 是将所有 Event 存储到数据库这种持久化存储中，目前只支持内嵌的 Derby</code>。</p>
<h3 id="Sink"><a href="#Sink" class="headerlink" title="Sink"></a>Sink</h3><p><code>Sink 不断地轮询 Channel 中的事件且批量地移除它们</code>，并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。</p>
<p>Sink 组件目的地包括 hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定义。</p>
<h3 id="Event"><a href="#Event" class="headerlink" title="Event"></a>Event</h3><p>传输单元，Flume 数据传输的基本单元，以 Event 的形式将数据从源头送至目的地。Event 由 <code>Header 和 Body</code> 两部分组成，Header 用来存放该 event 的一些属性，为K-V结构，Body 用来存放该条数据，形式为字节数组。</p>
<p><img src="/oct25-xxxxx/img/flume/flume02.png" alt="Event"></p>
<h1 id="Flume-快速入门"><a href="#Flume-快速入门" class="headerlink" title="Flume 快速入门"></a>Flume 快速入门</h1><h2 id="一、官网地址"><a href="#一、官网地址" class="headerlink" title="一、官网地址"></a>一、官网地址</h2><p>Flume 官网地址：<a target="_blank" rel="noopener" href="http://flume.apache.org/">http://flume.apache.org/</a></p>
<h2 id="二、Flume-安装部署"><a href="#二、Flume-安装部署" class="headerlink" title="二、Flume 安装部署"></a>二、Flume 安装部署</h2><h3 id="上传-tar-包，并解压"><a href="#上传-tar-包，并解压" class="headerlink" title="上传 tar 包，并解压"></a>上传 tar 包，并解压</h3><p>将 Jar 包上传到服务器，并解压</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">tar -zxvf apache-flume-1.11.0-bin.tar.gz -C /opt/module/</span><br></pre></td></tr></table></figure>

<p>重命名文件夹</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">mv</span> apache-flume-1.11.0-bin/ flume-1.11.0</span><br></pre></td></tr></table></figure>

<h3 id="删除依赖包"><a href="#删除依赖包" class="headerlink" title="删除依赖包"></a>删除依赖包</h3><p>进入到 Flume 根目录下的 lib&#x2F; 目录下</p>
<p>搜索并删除 guava-11.0.2.jar，以兼容 Hadoop-3.1.3</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#检索，看目录下是否有 guava 的包</span></span><br><span class="line">ll | grep guava</span><br><span class="line"></span><br><span class="line"><span class="comment">#删除</span></span><br><span class="line"><span class="built_in">rm</span> -rf guava-11.0.2.jar</span><br></pre></td></tr></table></figure>

<h3 id="修改配置文件"><a href="#修改配置文件" class="headerlink" title="修改配置文件"></a>修改配置文件</h3><p>进入到 Flume 根目录下的 conf&#x2F; 目录下</p>
<ol>
<li>修改 flume-env.sh 添加 Java 环境变量</li>
</ol>
<p>重命名配置文件</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cp</span> flume-env.sh.template flume-env.sh</span><br></pre></td></tr></table></figure>

<p>编辑配置文件，并修改以下项</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">export</span> JAVA_HOME=/opt/module/jdk1.8.0_351</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>修改日志生成路径</li>
</ol>
<p>编辑 log4j2.xml 配置文件</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">Properties</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;<span class="name">Property</span> <span class="attr">name</span>=<span class="string">&quot;LOG_DIR&quot;</span>&gt;</span>/opt/module/flume-1.11.0/logs<span class="tag">&lt;/<span class="name">Property</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">Properties</span>&gt;</span></span><br></pre></td></tr></table></figure>

<ol start="3">
<li>日志输出 console</li>
</ol>
<p>查看 log4j2.xml 配置文件，可以看到日志只输出到日志文件中，并没有输出到控制台，所以可以将 console 输出添加</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">Root</span> <span class="attr">level</span>=<span class="string">&quot;INFO&quot;</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">AppenderRef</span> <span class="attr">ref</span>=<span class="string">&quot;LogFile&quot;</span> /&gt;</span></span><br><span class="line">    <span class="comment">&lt;!-- 添加控制台输出 --&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">AppenderRef</span> <span class="attr">ref</span>=<span class="string">&quot;Console&quot;</span> /&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">Root</span>&gt;</span></span><br></pre></td></tr></table></figure>



<h1 id="Flume-入门案例"><a href="#Flume-入门案例" class="headerlink" title="Flume 入门案例"></a>Flume 入门案例</h1><h2 id="一、Flume-官方案例"><a href="#一、Flume-官方案例" class="headerlink" title="一、Flume 官方案例"></a>一、Flume 官方案例</h2><h3 id="案例需求"><a href="#案例需求" class="headerlink" title="案例需求"></a>案例需求</h3><p>使用 Flume 监听一个端口，采集该商品数据，并打印到控制台。</p>
<h3 id="需求分析"><a href="#需求分析" class="headerlink" title="需求分析"></a>需求分析</h3><p><img src="/oct25-xxxxx/img/flume/flume03.png" alt="官方案例图示"></p>
<ol>
<li>监控 Linux 端口数据（例如：44444 端口），使用 NetCat Source；</li>
<li>Channel 选择 Memory Channel；</li>
<li>将采集到的数据，使用 Logger Sink 输出；</li>
</ol>
<h3 id="准备工作"><a href="#准备工作" class="headerlink" title="准备工作"></a>准备工作</h3><p>需要安装一下 NetCat 工具</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo yum -y install nc</span><br></pre></td></tr></table></figure>

<p>为防止某个端口占用情况，可以提前查看44444端口是否被占用，如果被占用的话，可以换其他的端口号</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo netstat -nltp | grep 44444</span><br></pre></td></tr></table></figure>

<p>无输出即当前端口未被占用</p>
<p>测试是否可以正常使用</p>
<ol>
<li>启动服务端，监控44444端口数据</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc -lk 44444</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>启动客户端，往44444端口发送数据（需要使用新会话）</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc localhost 44444</span><br></pre></td></tr></table></figure>

<ol start="3">
<li>在客户端输入数据，看服务端是否可以接收到数据</li>
</ol>
<blockquote>
<p>注：当前还未使用 Flume，仅仅是看 nc 工具是否安装成功，并正常使用。</p>
</blockquote>
<h3 id="案例实现"><a href="#案例实现" class="headerlink" title="案例实现"></a>案例实现</h3><p>在任意目录下，创建一个存放 job 配置文件的文件夹</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">mkdir</span> job-sample</span><br></pre></td></tr></table></figure>

<p>创建官方案例 job 配置文件</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim flume-netcat-logger.conf</span><br></pre></td></tr></table></figure>

<p>从官网将案例粘贴下来</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># example.conf: A single-node Flume configuration</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义source/sink/channel名称</span></span><br><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义source类型</span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">netcat</span></span><br><span class="line"><span class="attr">a1.sources.r1.bind</span> = <span class="string">localhost</span></span><br><span class="line"><span class="attr">a1.sources.r1.port</span> = <span class="string">44444</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义sink类型</span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义channel类型</span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#绑定source和channel，以及sink和channel</span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>NetCat Source 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>source 类型，当前需配置为 netcat</td>
</tr>
<tr>
<td>bind</td>
<td>-</td>
<td>是</td>
<td>绑定的主机名或IP地址</td>
</tr>
<tr>
<td>port</td>
<td>-</td>
<td>是</td>
<td>绑定的端口号</td>
</tr>
</tbody></table>
<p>Memory Channel 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>channel 类型</td>
</tr>
<tr>
<td>capacity</td>
<td>100</td>
<td>否</td>
<td>channel 是存储的最大 Event 数量</td>
</tr>
<tr>
<td>transactionCapacity</td>
<td>100</td>
<td>否</td>
<td>从 source 获取，或者给 sink 的最大 Event 数量</td>
</tr>
</tbody></table>
<h3 id="启动测试"><a href="#启动测试" class="headerlink" title="启动测试"></a>启动测试</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -c ./conf/ -n a1 -f ./job_sample/flume-netcat-logger.conf</span><br></pre></td></tr></table></figure>

<p>参数说明：</p>
<ul>
<li>-c：指定配置文件目录</li>
<li>-n：当前 agent 名</li>
<li>-f：运行 job 配置文件名</li>
</ul>
<p>查看日志文件输出信息</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">tail</span> -100f flume.log</span><br></pre></td></tr></table></figure>

<p>日志文件中，输出以下日志，表示启动成功</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">Source starting</span><br><span class="line">Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]</span><br></pre></td></tr></table></figure>

<p>使用 nc 工具，往44444端口发送数据</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc localhost 44444</span><br></pre></td></tr></table></figure>

<figure class="highlight plaintext"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">hello world!</span><br></pre></td></tr></table></figure>

<p>再次查看日志输出</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">Source starting</span><br><span class="line">Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]</span><br><span class="line">Event: &#123; headers:&#123;&#125; body: 68 65 6C 6C 6F 20 77 6F 72 6C 64 21             hello world! &#125;</span><br></pre></td></tr></table></figure>

<h3 id="问题"><a href="#问题" class="headerlink" title="问题"></a>问题</h3><p>在 flume-1.11.0 版本中，执行下面命令启动 job</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -c ./conf/ -n a1 -f ./job_sample/flume-netcat-logger.conf</span><br></pre></td></tr></table></figure>

<p>在 job 启动后，并不会在控制台看到 <code>Source starting</code> 等启动成功的信息，控制台只打印了日志构架相关信息，如下所示：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line">Info: Sourcing environment configuration script /opt/module/flume-1.11.0/conf/flume-env.sh</span><br><span class="line">Info: Including Hadoop libraries found via (/opt/module/hadoop-3.3.4/bin/hadoop) <span class="keyword">for</span> HDFS access</span><br><span class="line">Info: Including Hive libraries found via () <span class="keyword">for</span> Hive access</span><br><span class="line">+ <span class="built_in">exec</span> /opt/module/jdk1.8.0_351/bin/java -Xmx20m -<span class="built_in">cp</span> <span class="string">&#x27;/opt/module/flume-1.11.0/conf:/opt/module/flume-1.11.0/lib/*:/opt/module/hadoop-3.3.4/etc/hadoop:/opt/module/hadoop-3.3.4/share/hadoop/common/lib/*:/opt/module/hadoop-3.3.4/share/hadoop/common/*:/opt/module/hadoop-3.3.4/share/hadoop/hdfs:/opt/module/hadoop-3.3.4/share/hadoop/hdfs/lib/*:/opt/module/hadoop-3.3.4/share/hadoop/hdfs/*:/opt/module/hadoop-3.3.4/share/hadoop/mapreduce/*:/opt/module/hadoop-3.3.4/share/hadoop/yarn:/opt/module/hadoop-3.3.4/share/hadoop/yarn/lib/*:/opt/module/hadoop-3.3.4/share/hadoop/yarn/*:/lib/*&#x27;</span> -Djava.library.path=:/opt/module/hadoop-3.3.4/lib/native org.apache.flume.node.Application -n a1 -f job_sample/flume-netcat-logger.conf</span><br><span class="line">SLF4J: Class path contains multiple SLF4J bindings.</span><br><span class="line">SLF4J: Found binding <span class="keyword">in</span> [jar:file:/opt/module/flume-1.11.0/lib/log4j-slf4j-impl-2.18.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]</span><br><span class="line">SLF4J: Found binding <span class="keyword">in</span> [jar:file:/opt/module/hadoop-3.3.4/share/hadoop/common/lib/slf4j-reload4j-1.7.36.jar!/org/slf4j/impl/StaticLoggerBinder.class]</span><br><span class="line">SLF4J: See http://www.slf4j.org/codes.html<span class="comment">#multiple_bindings for an explanation.</span></span><br><span class="line">SLF4J: Actual binding is of <span class="built_in">type</span> [org.apache.logging.slf4j.Log4jLoggerFactory]</span><br></pre></td></tr></table></figure>

<p>如何解决？</p>
<p>可以参考 <a href="#修改配置文件">修改配置文件</a> 小节，配置日志输出的目录，然后使用命令 <code>tail -100f flume.log</code> 监控目录文件即可</p>
<h2 id="二、实时监控本地文件到-HDFS"><a href="#二、实时监控本地文件到-HDFS" class="headerlink" title="二、实时监控本地文件到 HDFS"></a>二、实时监控本地文件到 HDFS</h2><h3 id="案例需求-1"><a href="#案例需求-1" class="headerlink" title="案例需求"></a>案例需求</h3><p>实时监控 Hive 日志文件，并将监测到的数据上传到 HDFS 中。</p>
<h3 id="需求分析-1"><a href="#需求分析-1" class="headerlink" title="需求分析"></a>需求分析</h3><p><img src="/oct25-xxxxx/img/flume/flume04.png" alt="Flume 图示"></p>
<ol>
<li>监控 Hive 日志文件目录，使用 Exec Source；</li>
<li>Channel 选择 Memory Channel；</li>
<li>将监控到的日志数据上传到 HDFS，直接使用 HDFS Sink；</li>
</ol>
<h3 id="准备工作-1"><a href="#准备工作-1" class="headerlink" title="准备工作"></a>准备工作</h3><p>需要部署 Hadoop 集群与 Hive，可以参考其他文档，此处部署过程略… …</p>
<p>Flume 要想将数据上传到 HDFS，需要提前添加 Hadoop 相关 jar 包</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">commons-configuration2-2.1.1.jar</span><br><span class="line">commons-io-2.8.0.jar</span><br><span class="line">hadoop-auth-3.3.4.jar</span><br><span class="line">hadoop-common-3.3.4.jar</span><br><span class="line">hadoop-hdfs-3.3.4.jar</span><br></pre></td></tr></table></figure>

<h3 id="案例实现-1"><a href="#案例实现-1" class="headerlink" title="案例实现"></a>案例实现</h3><p>创建案例配置文件</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim flume exec-hdfs.conf</span><br></pre></td></tr></table></figure>

<p>从官方文档找到对应配置粘贴并修改：</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">exec</span></span><br><span class="line"><span class="attr">a1.sources.r1.command</span> = <span class="string">tail -F /opt/module/hive-3.1.3/logs/hive.log</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">hdfs</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.path</span> = <span class="string">hdfs://hadoop102:8020/flume/%Y%m%d/%H%M</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.round</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.roundValue</span> = <span class="string">1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.roundUnit</span> = <span class="string">minute</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.filePrefix</span> = <span class="string">logs-</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.useLocalTimeStamp</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.batchSize</span> = <span class="string">100</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.fileType</span> = <span class="string">DataStream</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollInterval</span> = <span class="string">60</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollSize</span> = <span class="string">134217700</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollCount</span> = <span class="string">0</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>Exec Source 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>source 类型，当前需配置为 exec</td>
</tr>
<tr>
<td>command</td>
<td>-</td>
<td>需要执行的 Linux 命令</td>
</tr>
</tbody></table>
<p>HDFS Sink 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>sink 类型，当前需配置为 hdfs</td>
</tr>
<tr>
<td>hdfs.path</td>
<td>-</td>
<td>上传到 hdfs 路径</td>
</tr>
<tr>
<td>hdfs.round</td>
<td>false</td>
<td>是否按照时间滚动文件</td>
</tr>
<tr>
<td>hdfs.roundValue</td>
<td>1</td>
<td>多长时间滚动创建一个新的文件夹，单位由 hdfs.rundUnit 决定</td>
</tr>
<tr>
<td>hdfs.rundUnit</td>
<td>second</td>
<td>滚动时间单位</td>
</tr>
<tr>
<td>hdfs.filePrefix</td>
<td>FlumeData</td>
<td>上传文件前缀</td>
</tr>
<tr>
<td>hdfs.useLocalTimeStamp</td>
<td>false</td>
<td>是否使用本地时间戳</td>
</tr>
<tr>
<td>hdfs.batchSize</td>
<td>100</td>
<td>多少个event为一批次，flush 到 HDFS 上</td>
</tr>
<tr>
<td>hdfs.fileType</td>
<td>SequenceFile</td>
<td>文件格式，目前为 SequenceFile、DataStream、CompressedStream；<br/>（1)DataStream不会压缩输出文件；<br/>(2)CompressedStream需要配置 hdfs.codeC，具体参考官方文档</td>
</tr>
<tr>
<td>hdfs.rollInterval</td>
<td>30</td>
<td>等待多少秒滚动当前文件，单位秒</td>
</tr>
<tr>
<td>hdfs.rollSize</td>
<td>1024</td>
<td>触发滚动文件大小，单位字节</td>
</tr>
<tr>
<td>hdfs.rollCount</td>
<td>10</td>
<td>触发文件滚动event个数</td>
</tr>
</tbody></table>
<blockquote>
<p>Tips：</p>
<p>以下四个属性针对的是生成<strong>文件夹</strong>的策略</p>
<ul>
<li>a1.sinks.k1.hdfs.path &#x3D; hdfs:&#x2F;&#x2F;hadooop102:9000&#x2F;flume&#x2F;%Y%m%d&#x2F;%H%M：在hdfs上的路径名称</li>
<li>a1.sinks.k1.hdfs.round &#x3D; true：按照时间滚动文件夹</li>
<li>a1.sinks.k1.hdfs.roundValue &#x3D; 1：滚动的时间值</li>
<li>a1.sinks.k1.hdfs.roundUnit &#x3D; minute：滚动的时间单位</li>
</ul>
<p>以下三个属性针对的是生成<strong>文件</strong>的策略</p>
<ul>
<li>a2.sinks.k2.hdfs.rollInterval &#x3D; 60：60秒滚动生成一个新文件</li>
<li>a2.sinks.k2.hdfs.rollSize &#x3D; 134217700：128M滚动生成一个新文件</li>
<li>a2.sinks.k2.hdfs.rollCount &#x3D; 0：不采用event个数生成文件</li>
</ul>
</blockquote>
<h3 id="启动测试-1"><a href="#启动测试-1" class="headerlink" title="启动测试"></a>启动测试</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -c ./conf/ -n a1 -f ./job_sample/flume-exec-hdfs.conf</span><br></pre></td></tr></table></figure>

<p>启动并连接 Hive 测试，到 HDFS 上查看是否产生对应目录文件，等待一段时间后，看是否可以滚动生成对应目录</p>
<h2 id="三、实时监控指定目录下多个文件"><a href="#三、实时监控指定目录下多个文件" class="headerlink" title="三、实时监控指定目录下多个文件"></a>三、实时监控指定目录下多个文件</h2><h3 id="案例需求-2"><a href="#案例需求-2" class="headerlink" title="案例需求"></a>案例需求</h3><p>使用 Flume 监听指定的目录下面的所有文件，并上传到 HDFS。</p>
<h3 id="需求分析-2"><a href="#需求分析-2" class="headerlink" title="需求分析"></a>需求分析</h3><p><img src="/oct25-xxxxx/img/flume/flume05.png" alt="Flume 图示"></p>
<ol>
<li>监控指定目录下的文件，使用 Spooling Directory Source；</li>
<li>Channel 选择 Memory Channel；</li>
<li>将监控到的日志数据上传到 HDFS，直接使用 HDFS Sink；</li>
</ol>
<h3 id="案例实现-2"><a href="#案例实现-2" class="headerlink" title="案例实现"></a>案例实现</h3><p>创建案例配置文件</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim flume-spooldir-hdfs.conf</span><br></pre></td></tr></table></figure>

<p>从官方文档找到对应配置粘贴并修改：</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">spooldir</span></span><br><span class="line"><span class="attr">a1.sources.r1.spoolDir</span> = <span class="string">/opt/module/flume-1.11.0/upload</span></span><br><span class="line"><span class="attr">a1.sources.r1.fileSuffix</span> = <span class="string">.COMPLETED</span></span><br><span class="line"><span class="attr">a1.sources.r1.fileHeader</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sources.r1.ignorePattern</span> = <span class="string">([^ ]*\.tmp)</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">hdfs</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.path</span> = <span class="string">hdfs://hadoop102:8020/flume/%Y%m%d/%H%M</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.round</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.roundValue</span> = <span class="string">1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.roundUnit</span> = <span class="string">minute</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.filePrefix</span> = <span class="string">upload-</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.useLocalTimeStamp</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.batchSize</span> = <span class="string">100</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.fileType</span> = <span class="string">DataStream</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollInterval</span> = <span class="string">20</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollSize</span> = <span class="string">134217700</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollCount</span> = <span class="string">0</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>Spooling Directory Source 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>source 类型</td>
</tr>
<tr>
<td>spoolDir</td>
<td>-</td>
<td>是</td>
<td>要读取的目录</td>
</tr>
<tr>
<td>fireSuffix</td>
<td>.COMPLETED</td>
<td>否</td>
<td>给已读取文件请加的后缀</td>
</tr>
<tr>
<td>deletePolicy</td>
<td>never</td>
<td>否</td>
<td>是否删除已完成文件，never从不，immediate立即</td>
</tr>
<tr>
<td>fileHeader</td>
<td>false</td>
<td>否</td>
<td>是否将绝对路径文件名追加到 event 的 header 中</td>
</tr>
<tr>
<td>fileHederKey</td>
<td>file</td>
<td>否</td>
<td>将文件绝对路径文件名追加到 event 的 header 中的 key</td>
</tr>
<tr>
<td>includePattern</td>
<td>^.*$</td>
<td>否</td>
<td>指定要包含文件的正则表达式，如果同时匹配到 includePattern 和 ignorePattern 时，文件将被忽略</td>
</tr>
<tr>
<td>ignorePattern</td>
<td>^$</td>
<td>否</td>
<td>指定要忽略文件的正则表达式</td>
</tr>
</tbody></table>
<h3 id="启动测试-2"><a href="#启动测试-2" class="headerlink" title="启动测试"></a>启动测试</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -c ./conf/ -n a1 -f ./job_sample/flume-spooldir-hdfs.conf</span><br></pre></td></tr></table></figure>

<ol>
<li>将几个测试使用的文件添加到指定目录，看 HDFS 上是否有文件产生；</li>
<li>将后缀为 .tmp 结尾文件添加到指定目录，看是否会上传到 HDFS；</li>
<li>测试超过20秒后，看 HDFS 上是否会滚动生成一个新文件；</li>
<li>过1分钟后，看 HDFS 上是否会生成新的目录；</li>
</ol>
<blockquote>
<p>思考：假如将一个名为 1.COMPLETED 的文件添加到指定目录下，Flume 是否会上传到 HDFS？</p>
</blockquote>
<h2 id="四、实时监控目录下的多个追加文件"><a href="#四、实时监控目录下的多个追加文件" class="headerlink" title="四、实时监控目录下的多个追加文件"></a>四、实时监控目录下的多个追加文件</h2><h3 id="案例需求-3"><a href="#案例需求-3" class="headerlink" title="案例需求"></a>案例需求</h3><p>Exec Source 适用于监控一个实时追加的文件，<code>不能实现断点续传</code>；</p>
<p>Spooling Directory Source 适用于同步新文件，但是不适合对实时追加日志的文件进行进行监听并同步；</p>
<p>既想要监控多个目录下的文件，同时也想要监控目录中的文件，对追加内容进行同步，这时可以使用 Taildir Source 适合用于监听<code>多个实时追加的文件，并且能够实现断点续传</code></p>
<h3 id="需求分析-3"><a href="#需求分析-3" class="headerlink" title="需求分析"></a>需求分析</h3><p><img src="/oct25-xxxxx/img/flume/flume06.png" alt="Flume 图示"></p>
<ol>
<li>使用 Taildir Source 监听目录下的多个文件；</li>
<li>Channel 选择 Memory Channel；</li>
<li>将监控到的日志数据上传到 HDFS，直接使用 HDFS Sink；</li>
</ol>
<h3 id="案例实现-3"><a href="#案例实现-3" class="headerlink" title="案例实现"></a>案例实现</h3><p>创建案例配置文件</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim flume taildir-hdfs.conf</span><br></pre></td></tr></table></figure>

<p>从官方文档找到对应配置粘贴并修改：</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">TAILDIR</span></span><br><span class="line"><span class="attr">a1.sources.r1.positionFile</span> = <span class="string">/opt/module/flume-1.11.0/taildir_position.json</span></span><br><span class="line"><span class="attr">a1.sources.r1.filegroups</span> = <span class="string">f1 f2</span></span><br><span class="line"><span class="attr">a1.sources.r1.filegroups.f1</span> = <span class="string">/opt/module/flume-1.11.0/upload/dir1/.*file.*</span></span><br><span class="line"><span class="attr">a1.sources.r1.filegroups.f2</span> = <span class="string">/opt/module/flume-1.11.0/upload/dir2/.*log.*</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">hdfs</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.path</span> = <span class="string">hdfs://hadoop102:8020/flume/%Y%m%d/%H%M</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.round</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.roundValue</span> = <span class="string">1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.roundUnit</span> = <span class="string">minute</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.filePrefix</span> = <span class="string">upload2-</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.useLocalTimeStamp</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.batchSize</span> = <span class="string">100</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.fileType</span> = <span class="string">DataStream</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollInterval</span> = <span class="string">20</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollSize</span> = <span class="string">134217700</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hdfs.rollCount</span> = <span class="string">0</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>Taildir Source 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td>是</td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>source 类型</td>
</tr>
<tr>
<td>filegroups</td>
<td>-</td>
<td>是</td>
<td>使用空格分隔的文件组列表。每个文件组表示一个监听文件</td>
</tr>
<tr>
<td>filegroups.&lt;fileGroupName&gt;</td>
<td>-</td>
<td>是</td>
<td>文件组的绝对路径。正则表达式只用于文件名</td>
</tr>
<tr>
<td>positionFile</td>
<td>~&#x2F;.flume&#x2F;taildir_position.json</td>
<td>否</td>
<td>JSON格式文件，用于记录每个 inode，绝对路径和最后位置</td>
</tr>
</tbody></table>
<h3 id="启动测试-3"><a href="#启动测试-3" class="headerlink" title="启动测试"></a>启动测试</h3><figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -c ./conf/ -n a1 -f ./job_sample/flume-taildir-hdfs.conf</span><br></pre></td></tr></table></figure>

<p>测试往两个目录中添加指定文件，再测试添加不带关键字的文件，看 HDFS 上是否有对应的文件同步；</p>
<h3 id="问题-1"><a href="#问题-1" class="headerlink" title="问题"></a>问题</h3><h4 id="ⅰ问题说明"><a href="#ⅰ问题说明" class="headerlink" title="ⅰ问题说明"></a>ⅰ问题说明</h4><p>我们在配置 Taildir Source 时，会配置一个 JSON 文件，用于记录 inode、文件绝对路径和最后位置，Flume 会基于该文件实现断点续传功能。</p>
<p>我们进行过上面测试后，该文件会生成，可以查看下文件内容，如下所示：</p>
<figure class="highlight json"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="punctuation">[</span></span><br><span class="line">    <span class="punctuation">&#123;</span></span><br><span class="line">        <span class="attr">&quot;inode&quot;</span><span class="punctuation">:</span><span class="number">10450268</span><span class="punctuation">,</span></span><br><span class="line">        <span class="attr">&quot;pos&quot;</span><span class="punctuation">:</span><span class="number">10</span><span class="punctuation">,</span></span><br><span class="line">        <span class="attr">&quot;file&quot;</span><span class="punctuation">:</span><span class="string">&quot;/opt/module/flume-1.11.0/upload/dir1/123file.txt&quot;</span></span><br><span class="line">    <span class="punctuation">&#125;</span><span class="punctuation">,</span></span><br><span class="line">    <span class="punctuation">&#123;</span></span><br><span class="line">        <span class="attr">&quot;inode&quot;</span><span class="punctuation">:</span><span class="number">37358471</span><span class="punctuation">,</span></span><br><span class="line">        <span class="attr">&quot;pos&quot;</span><span class="punctuation">:</span><span class="number">10</span><span class="punctuation">,</span></span><br><span class="line">        <span class="attr">&quot;file&quot;</span><span class="punctuation">:</span><span class="string">&quot;/opt/module/flume-1.11.0/upload/dir2/789log.txt&quot;</span></span><br><span class="line">    <span class="punctuation">&#125;</span></span><br><span class="line"><span class="punctuation">]</span></span><br></pre></td></tr></table></figure>

<p>Flume 会使用 inode 与 file 字段作为联合主键去确定一个文件断点前的最后位置的，正常情况下，文件在不删除的情况下，inode 值是不会变的，所以一般如果文件名修改了，那联合主键发生变化了，就无法获取到对应位置了，从而无法实现断点续传了</p>
<p><strong>上述问题场景</strong>：监控日志文件目录，当日志框架使用的是 log4j 时，默认日志文件为 xxx.log，当到了每个零点时，日志文件滚动生成文件名为 xxx.log.yyyy-MM-dd 的日志文件，同时 xxx.log 重新保存新的日志数据。</p>
<h4 id="ⅱ问题复现"><a href="#ⅱ问题复现" class="headerlink" title="ⅱ问题复现"></a>ⅱ问题复现</h4><p>问题复现还是基于上面示例基础上，监听的目录为以下两个目录（测试只需要一个即可）：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">/opt/module/flume-1.11.0/upload/dir1/.*file.*</span><br><span class="line">/opt/module/flume-1.11.0/upload/dir2/.*<span class="built_in">log</span>.*</span><br></pre></td></tr></table></figure>

<ol>
<li>删除之前测试数据</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">rm</span> -rf /opt/module/flume-1.11.0/upload/dir1/*</span><br><span class="line"><span class="built_in">rm</span> -rf /opt/module/flume-1.11.0/upload/dir2/*</span><br><span class="line"></span><br><span class="line"><span class="built_in">rm</span> -rf /opt/module/flume-1.11.0/taildir_position.json</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>启动 job</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/flume-taildir-hdfs.conf</span><br></pre></td></tr></table></figure>

<ol start="3">
<li>创建测试文件，并追加数据</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">touch</span> ./upload/dir1/file1.txt</span><br><span class="line"><span class="built_in">echo</span> <span class="string">&quot;hello flume&quot;</span> &gt;&gt; ./upload/dir1/file1.txt</span><br><span class="line"><span class="built_in">echo</span> <span class="string">&quot;hello hadoop&quot;</span> &gt;&gt; ./upload/dir1/file1.txt</span><br></pre></td></tr></table></figure>

<ol start="4">
<li>查看 HDFS 上同步文件</li>
</ol>
<p>文件一</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">hello flume</span><br></pre></td></tr></table></figure>

<p>文件二</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">hello hadoop</span><br></pre></td></tr></table></figure>

<blockquote>
<p>注：有两个文件的原因是操作超过20s，上传时滚动生成的新文件</p>
</blockquote>
<ol start="5">
<li>修改文件名</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">mv</span> ./upload/dir1/file1.txt ./upload/dir1/file2.txt</span><br></pre></td></tr></table></figure>

<ol start="6">
<li>再次查看 HDFS 上同步文件</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">hello flume</span><br><span class="line">hello hadoop</span><br></pre></td></tr></table></figure>

<blockquote>
<p>说明：将文件 file2.txt 重新上传，并且验证修改文件名无法实现断点续传。</p>
</blockquote>
<h4 id="ⅲ-修改源码解决"><a href="#ⅲ-修改源码解决" class="headerlink" title="ⅲ 修改源码解决"></a>ⅲ 修改源码解决</h4><ol>
<li>从官网下载源码，解压，并使用 IDEA 工具打开；</li>
<li>找到 flume-ng-sources&#x2F;flume-taildir-source；</li>
<li>修改 TailFile 代码（大约125行左右）</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="type">boolean</span> <span class="title function_">updatePos</span><span class="params">(String path, <span class="type">long</span> inode, <span class="type">long</span> pos)</span> <span class="keyword">throws</span> IOException &#123;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 由inode和文件路径作为联合主键</span></span><br><span class="line"><span class="comment">     * 如果文件路径（主要指文件名）发生变化时，这样就无法找到文件最后的位置进行断点续传了，就会出现将整个文件重新上传的情况</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line"><span class="comment">//    if (this.inode == inode &amp;&amp; this.path.equals(path)) &#123;</span></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 优化后，只使用inode作为主键</span></span><br><span class="line"><span class="comment">     * 这样就是无论文件路径（主要指文件名）怎么变化，都可以通过inode找到对应位置文件，从而定位到该文件最后位置断点续传</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">if</span> (<span class="built_in">this</span>.inode == inode) &#123;</span><br><span class="line">        </span><br><span class="line">        setPos(pos);</span><br><span class="line">        updateFilePos(pos);</span><br><span class="line">        logger.info(<span class="string">&quot;Updated position, file: &quot;</span> + path + <span class="string">&quot;, inode: &quot;</span> + inode + <span class="string">&quot;, pos: &quot;</span> + pos);</span><br><span class="line">        <span class="keyword">return</span> <span class="literal">true</span>;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">return</span> <span class="literal">false</span>;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="4">
<li>修改 ReliableTaildirEventReader 代码（大约255行左右）</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line"><span class="type">TailFile</span> <span class="variable">tf</span> <span class="operator">=</span> tailFiles.get(inode);</span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * tf是通过inode获取的TailFile，包含inode、最后位置pos、文件绝对路径file</span></span><br><span class="line"><span class="comment"> * 1、判断tf为空，表示inode不存在</span></span><br><span class="line"><span class="comment"> * 2、同时，tf中获取的路径，与f文件的绝对路径不相同</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="comment">// if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) &#123;</span></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * 优化后，只判断tf为空即可</span></span><br><span class="line"><span class="comment"> * 表示通过inode获取到的tf非空，即执行 else 分支</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">if</span> (tf == <span class="literal">null</span>) &#123;</span><br><span class="line">    </span><br><span class="line">    <span class="type">long</span> <span class="variable">startPos</span> <span class="operator">=</span> skipToEnd ? f.length() : <span class="number">0</span>;</span><br><span class="line">    tf = openFile(f, headers, inode, startPos);</span><br><span class="line">&#125; <span class="keyword">else</span> &#123;</span><br><span class="line">    <span class="type">boolean</span> <span class="variable">updated</span> <span class="operator">=</span> tf.getLastUpdated() &lt; f.lastModified() || tf.getPos() != f.length();</span><br><span class="line">    <span class="keyword">if</span> (updated) &#123;</span><br><span class="line">        <span class="keyword">if</span> (tf.getRaf() == <span class="literal">null</span>) &#123;</span><br><span class="line">            tf = openFile(f, headers, inode, tf.getPos());</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (f.length() &lt; tf.getPos()) &#123;</span><br><span class="line">            logger.info(<span class="string">&quot;Pos &quot;</span> + tf.getPos() + <span class="string">&quot; is larger than file size! &quot;</span></span><br><span class="line">                        + <span class="string">&quot;Restarting from pos 0, file: &quot;</span> + tf.getPath() + <span class="string">&quot;, inode: &quot;</span> + inode);</span><br><span class="line">            tf.updatePos(tf.getPath(), inode, <span class="number">0</span>);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    tf.setNeedTail(updated);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="5">
<li>重新打包</li>
<li>将打好的包，放到 flume-1.11.0&#x2F;lib&#x2F; 目录下</li>
<li>重启完成上面测试<ul>
<li>看是否还有新文件产生，并且是所有数据</li>
<li>往修改名字后的文件（file2）中追加数据，看能否实现断点续传</li>
</ul>
</li>
</ol>
<p>如果，在修改文件名后，HDFS 上生成的文件中只有追加数据，那说明修改成功，并解决上面问题。</p>
<h1 id="Flume-进阶"><a href="#Flume-进阶" class="headerlink" title="Flume 进阶"></a>Flume 进阶</h1><h2 id="一、Flume-事务"><a href="#一、Flume-事务" class="headerlink" title="一、Flume 事务"></a>一、Flume 事务</h2><p><img src="/oct25-xxxxx/img/flume/flume07.png" alt="Flume 事务"></p>
<p>Put 事务流程：</p>
<ul>
<li>doPut：将批数据先写入临时缓冲区 putList；</li>
<li>doCommit：检查 channel 内存队列是否足够合并；</li>
<li>doRollback：channel 内存队列空间不足，回滚数据；</li>
</ul>
<p>Take 事务流程：</p>
<ul>
<li>doTake：将数据取到临时缓冲区 takeList，并将数据发送到 HDFS；</li>
<li>doCommit：如果数据全部发送成功，则清除临时缓冲区 takeList；</li>
<li>doRollback：数据发送过程中，如果出现异常，rollback 将临时缓冲区 takeList 中的数据归还给 channel 内存队列；</li>
</ul>
<h2 id="二、Flume-Agent-内部原理"><a href="#二、Flume-Agent-内部原理" class="headerlink" title="二、Flume Agent 内部原理"></a>二、Flume Agent 内部原理</h2><p><img src="/oct25-xxxxx/img/flume/flume08.png" alt="Flume Agent 内部原理"></p>
<h2 id="三、Flume-拓扑结构"><a href="#三、Flume-拓扑结构" class="headerlink" title="三、Flume 拓扑结构"></a>三、Flume 拓扑结构</h2><h3 id="简单串联"><a href="#简单串联" class="headerlink" title="简单串联"></a>简单串联</h3><p><img src="/oct25-xxxxx/img/flume/flume09.png" alt="简单串联"></p>
<p>这种模式是将多个 Flume 顺序连接起来，从最初的 Source 开始到最终的 Sink 传送的目的存储系统。此模式不建议桥接过多的 Flume 数量，Flume 数量过多不仅会影响传输速率，而且一旦传输过程中某个节点 Flume 宕机，会影响整个传输系统。</p>
<h3 id="复制和多路复用"><a href="#复制和多路复用" class="headerlink" title="复制和多路复用"></a>复制和多路复用</h3><p><img src="/oct25-xxxxx/img/flume/flume10.png" alt="复制和多路复用"></p>
<h3 id="负载均衡和故障转移"><a href="#负载均衡和故障转移" class="headerlink" title="负载均衡和故障转移"></a>负载均衡和故障转移</h3><p><img src="/oct25-xxxxx/img/flume/flume11.png" alt="负载均衡和故障转移"></p>
<h2 id="四、Flume-企业开发案例"><a href="#四、Flume-企业开发案例" class="headerlink" title="四、Flume 企业开发案例"></a>四、Flume 企业开发案例</h2><h3 id="复制和多路复用-1"><a href="#复制和多路复用-1" class="headerlink" title="复制和多路复用"></a>复制和多路复用</h3><p><img src="/oct25-xxxxx/img/flume/flume12.png" alt="复制和多路复用"></p>
<h4 id="ⅰ案例需求"><a href="#ⅰ案例需求" class="headerlink" title="ⅰ案例需求"></a>ⅰ案例需求</h4><ol>
<li><p>使用 Flume-1 监控文件变动，可以使用 Exec Source 或者 Taildir Source；</p>
</li>
<li><p>将 Flume-1 同步的数据，同时传递给 Flume-2 和 Flume-3；</p>
</li>
<li><p>Flume-2 负责将数据同步到 HDFS；</p>
</li>
<li><p>Flume-3 负责将数据同步到日志文件中；</p>
</li>
</ol>
<h4 id="ⅱ案例实现"><a href="#ⅱ案例实现" class="headerlink" title="ⅱ案例实现"></a>ⅱ案例实现</h4><p>配置文件统一放到 <code>/opt/module/flume-1.11.0/job-sample/sample1/</code> 目录下。</p>
<p>创建 exec-flume.conf，下游连接两个 Flume</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1 k2</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1 c2</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">exec</span></span><br><span class="line"><span class="attr">a1.sources.r1.command</span> = <span class="string">tail -F /opt/module/hive-3.1.3/logs/hive.log</span></span><br><span class="line"><span class="attr">a1.sources.r1.shell</span> = <span class="string">/bin/bash -c</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"></span><br><span class="line"><span class="attr">a1.channels.c2.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c2.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c2.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hostname</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a1.sinks.k1.port</span> = <span class="string">4041</span></span><br><span class="line"></span><br><span class="line"><span class="attr">a1.sinks.k2.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k2.hostname</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a1.sinks.k2.port</span> = <span class="string">4042</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1 c2</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k2.channel</span> = <span class="string">c2</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-hdfs.conf，将上游数据发送到 HDFS </p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a2.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a2.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a2.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a2.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a2.sources.r1.bind</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a2.sources.r1.port</span> = <span class="string">4041</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a2.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a2.sinks.k1.type</span> = <span class="string">hdfs</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.path</span> = <span class="string">hdfs://hadoop102:8020/flume2/%Y%m%d/%H%M</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.round</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.roundValue</span> = <span class="string">1</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.roundUnit</span> = <span class="string">minute</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.filePrefix</span> = <span class="string">sample1-</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.useLocalTimeStamp</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.batchSize</span> = <span class="string">100</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.fileType</span> = <span class="string">DataStream</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.rollInterval</span> = <span class="string">600</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.rollSize</span> = <span class="string">134217700</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hdfs.rollCount</span> = <span class="string">0</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a2.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a2.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-file.conf，将上游数据保存本地文件</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a3.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a3.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a3.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a3.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a3.sources.r1.bind</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a3.sources.r1.port</span> = <span class="string">4042</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a3.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a3.sinks.k1.type</span> = <span class="string">file_roll</span></span><br><span class="line"><span class="attr">a3.sinks.k1.sink.directory</span> = <span class="string">/opt/module/flume-1.11.0/output/</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a3.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a3.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>Avro Sink 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>sink 类型</td>
</tr>
<tr>
<td>hostname</td>
<td>-</td>
<td>是</td>
<td>连接主机名或IP地址</td>
</tr>
<tr>
<td>port</td>
<td>-</td>
<td>是</td>
<td>监听端口</td>
</tr>
</tbody></table>
<p>Avro Source 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>sink 类型</td>
</tr>
<tr>
<td>bind</td>
<td>-</td>
<td>是</td>
<td>绑定主机名或IP地址</td>
</tr>
<tr>
<td>port</td>
<td>-</td>
<td>是</td>
<td>监听端口</td>
</tr>
</tbody></table>
<p>File Roll Sink 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>channel</td>
<td>-</td>
<td></td>
<td></td>
</tr>
<tr>
<td>type</td>
<td>-</td>
<td>是</td>
<td>sink 类型</td>
</tr>
<tr>
<td>sink.directory</td>
<td>-</td>
<td>是</td>
<td>文件存储的目录</td>
</tr>
</tbody></table>
<h4 id="ⅲ启动测试"><a href="#ⅲ启动测试" class="headerlink" title="ⅲ启动测试"></a>ⅲ启动测试</h4><p>启动 flume-file.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a3 -c conf/ -f job_sample/sample1/flume-file.conf</span><br></pre></td></tr></table></figure>

<p>启动 flume-hdfs.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a2 -c conf/ -f job_sample/sample1/flume-hdfs.conf</span><br></pre></td></tr></table></figure>

<p>启动 exec-flume.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/sample1/exec-flume.conf</span><br></pre></td></tr></table></figure>

<blockquote>
<p>需要注意一下</p>
<ol>
<li>三个 job 的启动顺序，需要先着后面的 a2 a3 启动，才可以启动 a1；如果先将 a1 启动会报错：hadoop102:4041 和 4042 拒绝连接。</li>
<li>a2 需要使用到 HDFS，a1 需要监控 Hive 日志，所以需要将 HDFS 和 Hive 服务提前启动起来。</li>
</ol>
</blockquote>
<p>启动正常，查看 Flume 看是否有其他报错</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">tail</span> -F /opt/module/flume-1.11.0/log/flume.log</span><br></pre></td></tr></table></figure>

<p>如果正常的话，连接 Hive 执行一些操作后，看本地目录与 HDFS 上是否有对应数据产生。</p>
<h3 id="负载均衡和故障转移-1"><a href="#负载均衡和故障转移-1" class="headerlink" title="负载均衡和故障转移"></a>负载均衡和故障转移</h3><p><img src="/oct25-xxxxx/img/flume/flume13.png" alt="负载均衡和故障转移"></p>
<h4 id="ⅰ案例需求-1"><a href="#ⅰ案例需求-1" class="headerlink" title="ⅰ案例需求"></a>ⅰ案例需求</h4><ol>
<li>Flume-1 使用 NetCat Source 监控服务器端口数据，并将数据同步到 Flume-2 和 Flume-3；</li>
<li>Flume-2 和 Flume-3 采用 FailoverSinkProcessor，实现故障转移；</li>
<li>Flume-2 和 Flume-3 都将数据输出到控制台即可；</li>
</ol>
<h4 id="ⅱ案例实现-1"><a href="#ⅱ案例实现-1" class="headerlink" title="ⅱ案例实现"></a>ⅱ案例实现</h4><p>配置文件统一放到 <code>/opt/module/flume-1.11.0/job-sample/sample2/</code> 目录下。</p>
<p>创建 netcat-flume.conf，监控端口数据</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1 k2</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinkgroups</span> = <span class="string">g1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">netcat</span></span><br><span class="line"><span class="attr">a1.sources.r1.bind</span> = <span class="string">localhost</span></span><br><span class="line"><span class="attr">a1.sources.r1.port</span> = <span class="string">44444</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hostname</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a1.sinks.k1.port</span> = <span class="string">4041</span></span><br><span class="line"></span><br><span class="line"><span class="attr">a1.sinks.k2.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k2.hostname</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a1.sinks.k2.port</span> = <span class="string">4042</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink processor</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.type</span> = <span class="string">failover</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.priority.k1</span> = <span class="string">5 # 优先级</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.priority.k2</span> = <span class="string">10</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.maxpenalty</span> = <span class="string">10000</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.sinks</span> = <span class="string">k1 k2</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k2.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-log.conf，将数据输出到控制台</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a2.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a2.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a2.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a2.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a2.sources.r1.bind</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a2.sources.r1.port</span> = <span class="string">4041</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a2.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a2.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a2.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a2.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-log2.conf，将数据输出到控制台</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a3.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a3.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a3.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a3.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a3.sources.r1.bind</span> = <span class="string">hadoop102</span></span><br><span class="line"><span class="attr">a3.sources.r1.port</span> = <span class="string">4042</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a3.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a3.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a3.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a3.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>Failover Sink Processor 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>sinks</td>
<td>-</td>
<td>是</td>
<td>多个逗号分隔的 sink</td>
</tr>
<tr>
<td>processor.type</td>
<td>default</td>
<td>是</td>
<td>sink 处理器类型，当前为  failover</td>
</tr>
<tr>
<td>processor.priority.&lt;sinkName&gt;</td>
<td>-</td>
<td>是</td>
<td>优先级，绝对值越大，表示优先级越高</td>
</tr>
<tr>
<td>processor.maxpenalty</td>
<td>30000</td>
<td>否</td>
<td>故障接收器的最大回退周期，单位：毫秒</td>
</tr>
</tbody></table>
<h4 id="ⅲ启动测试-1"><a href="#ⅲ启动测试-1" class="headerlink" title="ⅲ启动测试"></a>ⅲ启动测试</h4><p>启动 flume-logger.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a2 -c conf/ -f job_sample/sample2/flume-logger.conf</span><br></pre></td></tr></table></figure>

<p>启动 flume-log2.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a3 -c conf/ -f job_sample/sample2/flume-logger.conf</span><br></pre></td></tr></table></figure>

<p>启动 netcat-flume.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/sample2/flume-logger.conf</span><br></pre></td></tr></table></figure>

<p>启动正常，查看 Flume 看是否有其他报错</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">tail</span> -F /opt/module/flume-1.11.0/log/flume.log</span><br></pre></td></tr></table></figure>

<p>使用 netcat 工具往44444端口输入数据</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc localhost 44444</span><br></pre></td></tr></table></figure>

<p>查看日志文件是否有数据输出</p>
<p>将 flume-logger.conf 停止，看日志是否会报错，并且再次输入数据，看控制台是否还会打印出数据。</p>
<h4 id="ⅳ负载均衡"><a href="#ⅳ负载均衡" class="headerlink" title="ⅳ负载均衡"></a>ⅳ负载均衡</h4><p>刚刚案例中，sink 的处理器类型配置的是 failover 故障转移，要想修改为负载均衡策略，只需要将 netcat-flume.conf 中的 sinkgroups 部分进行修改</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Describe the sink processor</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.type</span> = <span class="string">failover</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.priority.k1</span> = <span class="string">5 # 优先级</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.priority.k2</span> = <span class="string">10</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.maxpenalty</span> = <span class="string">10000</span></span><br></pre></td></tr></table></figure>

<p>修改为下面负载均衡配置即可：</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Describe the sink processor</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.type</span> = <span class="string">load_balance</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.backoff</span> = <span class="string">true</span></span><br><span class="line"><span class="attr">a1.sinkgroups.g1.processor.selector</span> = <span class="string">random</span></span><br></pre></td></tr></table></figure>

<p>load balancing Sink Processor 配置说明：</p>
<table>
<thead>
<tr>
<th>name</th>
<th>default</th>
<th>required</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>sinks</td>
<td>-</td>
<td>是</td>
<td>多个逗号分隔的 sink</td>
</tr>
<tr>
<td>processor.type</td>
<td>default</td>
<td>是</td>
<td>sink 处理器类型，当前为 load_balance</td>
</tr>
<tr>
<td>processor.backoff</td>
<td>false</td>
<td>否</td>
<td>回退机制</td>
</tr>
<tr>
<td>processor.selector</td>
<td>round_robin</td>
<td>否</td>
<td>负载均衡机制，必须从 AbstractSinkSelector 继承的自定义类的 round_robin、random 或者 FQCN 选择</td>
</tr>
</tbody></table>
<blockquote>
<p>注意：</p>
<p>processor.selector 配置为 round_robin 后再次测试时，会发现两个 sink 并不是轮询收到数据的，可能会出现一种情况，就是一个 sink 数据多，另一个很少，看来并不像是轮询机制。</p>
<p>这是说明下，轮询并不是轮询发数据，而是 sink 组中的 sink 轮询去取，这样就会出现上面的情况，sink 轮询取数据时，轮到某个 sink 时，刚好没数据，这时就轮空了，不会等着，而是接收同其他 sink 继续轮询，这样就会导致 某个 sink 数据少的情况。</p>
</blockquote>
<h3 id="聚合"><a href="#聚合" class="headerlink" title="聚合"></a>聚合</h3><p><img src="/oct25-xxxxx/img/flume/flume14.png" alt="聚合"></p>
<h4 id="ⅰ案例需求-2"><a href="#ⅰ案例需求-2" class="headerlink" title="ⅰ案例需求"></a>ⅰ案例需求</h4><p>这将使用三台机器测试</p>
<ol>
<li>hadoop102，作为 Flume-1 使用 Exec Source 监控日志文件 &#x2F;opt&#x2F;module&#x2F;hive-3.1.3&#x2F;logs&#x2F;hive.log；</li>
<li>hadoop103，作为 Flume-2 使用 NetCat Source 监控端口数据；</li>
<li>hadoop104，作为服务端 Flume-3，将 Flume-1 和 Flume-2 都将数据同步给 Flume-3，由 Flume-3 将数据打印到控制台；</li>
</ol>
<h4 id="ⅱ案例实现-2"><a href="#ⅱ案例实现-2" class="headerlink" title="ⅱ案例实现"></a>ⅱ案例实现</h4><p>配置文件统一放到 <code>/opt/module/flume-1.11.0/job-sample/sample3/</code> 目录下。</p>
<p>创建 exec-flume.conf，监控 Hive 日志文件</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">exec</span></span><br><span class="line"><span class="attr">a1.sources.r1.command</span> = <span class="string">tail -F /opt/module/hive-3.1.3/logs/hive.log</span></span><br><span class="line"><span class="attr">a1.sources.r1.shell</span> = <span class="string">/bin/bash -c</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hostname</span> = <span class="string">hadoop104</span></span><br><span class="line"><span class="attr">a1.sinks.k1.port</span> = <span class="string">4040</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>创建 netcat-flume.conf，监控端口</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a2.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a2.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a2.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a2.sources.r1.type</span> = <span class="string">netcat</span></span><br><span class="line"><span class="attr">a2.sources.r1.bind</span> = <span class="string">localhost</span></span><br><span class="line"><span class="attr">a2.sources.r1.port</span> = <span class="string">44444</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a2.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a2.sinks.k1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a2.sinks.k1.hostname</span> = <span class="string">hadoop104</span></span><br><span class="line"><span class="attr">a2.sinks.k1.port</span> = <span class="string">4040</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a2.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a2.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-log.conf，将上游数据打印到控制台</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a3.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a3.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a3.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a3.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a3.sources.r1.bind</span> = <span class="string">hadoop104</span></span><br><span class="line"><span class="attr">a3.sources.r1.port</span> = <span class="string">4040</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a3.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a3.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a3.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a3.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>由于测试需要多台服务器，所以需要将 Flume 分发到其他机器。</p>
<h4 id="ⅲ启动测试-2"><a href="#ⅲ启动测试-2" class="headerlink" title="ⅲ启动测试"></a>ⅲ启动测试</h4><p>由于监听的是 Hive 日志文件，所以需要提前将 Hadoop 集群与Hive启动，然后再启动 job 测试。</p>
<p>在 hadoop104，启动 fluem-log.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a3 -c conf/ -f job_sample/sample3/fluem-log.conf</span><br></pre></td></tr></table></figure>

<p>在 hadoop103，启动 netcat-flume.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a2 -c conf/ -f job_sample/sample3/netcat-flume.conf</span><br></pre></td></tr></table></figure>

<p>在 hadoop102，启动 exec-flume.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/sample3/exec-flume.conf</span><br></pre></td></tr></table></figure>

<p>启动正常，查看 Flume 看是否有其他报错</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">tail</span> -F /opt/module/flume-1.11.0/log/flume.log</span><br></pre></td></tr></table></figure>

<p>在 hadoop102 上连接 Hive，并进行一些操作产生日志数据</p>
<p>在 hadoop103 上使用 <code>nc localhost 44444</code> 命令往端口发送数据</p>
<p>在 hadoop104 上查看 Flume 日志，看数据是否同步到 hadoop104</p>
<h2 id="五、自定义拦截器-Interceptor"><a href="#五、自定义拦截器-Interceptor" class="headerlink" title="五、自定义拦截器 Interceptor"></a>五、自定义拦截器 Interceptor</h2><h3 id="前言"><a href="#前言" class="headerlink" title="前言"></a>前言</h3><p>结合 <a href="#二、Flume-Agent-内部原理">二、Flume Agent 内部原理</a> 章节的图示</p>
<ul>
<li>Source 接收到数据后，会由 Channel Processor 进行处理；</li>
<li>Channel Processor 会将 Event 传递给拦截器链；</li>
<li>经过拦截器后，再由 Channel Selector 处理。Channel Selector 默认类型为 replicating，就是将数据发往所有 Channel，案例 <a href="#复制和多路复用">复制和多路复用</a> 默认使用的默认的 Channel Selector；</li>
<li>Channel Processor 会根据 Channel Selector 的结果，将数据发往不同 Channel；</li>
</ul>
<h3 id="案例需求-4"><a href="#案例需求-4" class="headerlink" title="案例需求"></a>案例需求</h3><p>使用 Flume 采集服务器本地日志，需要按照日志类型的不同，将不同各类的日志发往不同系统。</p>
<p>以端口数据模拟日志，通过以字母开头（首字母为 A-Z 或 a-z）模拟不同类型的日志，需要通过自定义 interceptor 区分是否以字母开头，从而将数据发往不同系统（Channel）。</p>
<p><img src="/oct25-xxxxx/img/flume/flume15.png" alt="自定义interceptor和Multiplexing Channel Selector"></p>
<h3 id="需求分析-4"><a href="#需求分析-4" class="headerlink" title="需求分析"></a>需求分析</h3><ol>
<li>自定义 Interceptor，并且实现 Flume 中的 Interceptor 接口；<ul>
<li>获取 Event 中的数据，并且根据需求判断</li>
<li>根据判断结果，分别在 header 中添加不同的标识</li>
</ul>
</li>
<li>在创建 netcat-flume.conf 时，需要定义 Channel Select，并对上一步骤添加的标识，将 Event 分别发往不同 Channel 中；</li>
<li>下游的两个 flume-log.conf 接收到数据直接打印到控制台即可；</li>
</ol>
<h3 id="案例实现-4"><a href="#案例实现-4" class="headerlink" title="案例实现"></a>案例实现</h3><h4 id="ⅰ自定义-Interceptor"><a href="#ⅰ自定义-Interceptor" class="headerlink" title="ⅰ自定义 Interceptor"></a>ⅰ自定义 Interceptor</h4><ol>
<li>创建 Maven 工程</li>
</ol>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.example<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>FlumeDemo<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">version</span>&gt;</span>1.0-SNAPSHOT<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br></pre></td></tr></table></figure>

<ol start="2">
<li>导入 Flume 依赖</li>
</ol>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.flume<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>flume-ng-core<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.11.0<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<ol start="3">
<li>自定义 Interceptor，并实现 Interceptor 接口</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> org.example.interceptor;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> org.apache.flume.Context;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.Event;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.interceptor.Interceptor;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.nio.charset.StandardCharsets;</span><br><span class="line"><span class="keyword">import</span> java.util.List;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@Description</span> TODO 自定义 interceptor</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * 需求：检查数据是否以字母开头</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">TypeInterceptor</span> <span class="keyword">implements</span> <span class="title class_">Interceptor</span> &#123;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 初始化</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">initialize</span><span class="params">()</span> &#123;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理单个 Event 方法</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> event</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> Event <span class="title function_">intercept</span><span class="params">(Event event)</span> &#123;</span><br><span class="line">        <span class="comment">//获取body，并且转换为字符串，获取第一个字符</span></span><br><span class="line">        <span class="type">String</span> <span class="variable">body</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">String</span>(event.getBody(), StandardCharsets.UTF_8);</span><br><span class="line">        <span class="type">char</span> <span class="variable">firstChar</span> <span class="operator">=</span> body.charAt(<span class="number">0</span>);</span><br><span class="line"></span><br><span class="line">        <span class="comment">//判断是否是字母</span></span><br><span class="line">        <span class="keyword">if</span>(Character.isLetter(firstChar)) &#123;</span><br><span class="line">            event.getHeaders().put(<span class="string">&quot;type&quot;</span>, <span class="string">&quot;alphabet&quot;</span>);</span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">            event.getHeaders().put(<span class="string">&quot;type&quot;</span>, <span class="string">&quot;other&quot;</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> event;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理一组 Event 方法</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> events</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> List&lt;Event&gt; <span class="title function_">intercept</span><span class="params">(List&lt;Event&gt; events)</span> &#123;</span><br><span class="line">        events.forEach(<span class="built_in">this</span>::intercept);</span><br><span class="line">        <span class="keyword">return</span> events;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 关闭资源</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">close</span><span class="params">()</span> &#123;</span><br><span class="line"></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">class</span> <span class="title class_">Builder</span> <span class="keyword">implements</span> <span class="title class_">Interceptor</span>.Builder &#123;</span><br><span class="line"></span><br><span class="line">        <span class="comment">/**</span></span><br><span class="line"><span class="comment">         * 构建拦截器</span></span><br><span class="line"><span class="comment">         *</span></span><br><span class="line"><span class="comment">         * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">         */</span></span><br><span class="line">        <span class="meta">@Override</span></span><br><span class="line">        <span class="keyword">public</span> Interceptor <span class="title function_">build</span><span class="params">()</span> &#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">new</span> <span class="title class_">TypeInterceptor</span>();</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="comment">/**</span></span><br><span class="line"><span class="comment">         * 通过配置文件对 Builder 进行配置</span></span><br><span class="line"><span class="comment">         *</span></span><br><span class="line"><span class="comment">         * <span class="doctag">@param</span> context</span></span><br><span class="line"><span class="comment">         */</span></span><br><span class="line">        <span class="meta">@Override</span></span><br><span class="line">        <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">configure</span><span class="params">(Context context)</span> &#123;</span><br><span class="line"></span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="4">
<li>将完成后的代码打包，并且上传到 <code>/opt/module/flume-1.11.0/lib/</code> 目录下</li>
</ol>
<h4 id="ⅱ编写Job配置文件"><a href="#ⅱ编写Job配置文件" class="headerlink" title="ⅱ编写Job配置文件"></a>ⅱ编写Job配置文件</h4><p>配置文件统一放到 <code>/opt/module/flume-1.11.0/job-sample/interceptor/</code> 目录下。</p>
<p>创建 netcat-flume.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1 k2</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1 c2</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">netcat</span></span><br><span class="line"><span class="attr">a1.sources.r1.bind</span> = <span class="string">localhost</span></span><br><span class="line"><span class="attr">a1.sources.r1.port</span> = <span class="string">44444</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the interceptors</span></span><br><span class="line"><span class="attr">a1.sources.r1.interceptors</span> = <span class="string">i1</span></span><br><span class="line"><span class="attr">a1.sources.r1.interceptors.i1.type</span> = <span class="string">org.example.interceptor.TypeInterceptor$Builder</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the Channel Selector</span></span><br><span class="line"><span class="attr">a1.sources.r1.selector.type</span> = <span class="string">multiplexing</span></span><br><span class="line"><span class="attr">a1.sources.r1.selector.header</span> = <span class="string">type</span></span><br><span class="line"><span class="attr">a1.sources.r1.selector.mapping.alphabet</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sources.r1.selector.mapping.other</span> = <span class="string">c2</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"></span><br><span class="line"><span class="attr">a1.channels.c2.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c2.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c2.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k1.hostname</span> = <span class="string">hadoop103</span></span><br><span class="line"><span class="attr">a1.sinks.k1.port</span> = <span class="string">4041</span></span><br><span class="line"></span><br><span class="line"><span class="attr">a1.sinks.k2.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a1.sinks.k2.hostname</span> = <span class="string">hadoop104</span></span><br><span class="line"><span class="attr">a1.sinks.k2.port</span> = <span class="string">4042</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1 c2</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k2.channel</span> = <span class="string">c2</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-log.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a2.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a2.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a2.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a2.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a2.sources.r1.bind</span> = <span class="string">hadoop103</span></span><br><span class="line"><span class="attr">a2.sources.r1.port</span> = <span class="string">4040</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a2.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a2.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a2.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a2.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a2.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<p>创建 flume-log2.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a3.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a3.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a3.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a3.sources.r1.type</span> = <span class="string">avro</span></span><br><span class="line"><span class="attr">a3.sources.r1.bind</span> = <span class="string">hadoop104</span></span><br><span class="line"><span class="attr">a3.sources.r1.port</span> = <span class="string">4041</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a3.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a3.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a3.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a3.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a3.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<h3 id="启动测试-4"><a href="#启动测试-4" class="headerlink" title="启动测试"></a>启动测试</h3><p>在 hadoop103，启动 flume-log.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a2 -c conf/ -f job_sample/interceptor/flume-log.conf</span><br></pre></td></tr></table></figure>

<p>在 hadoop104，启动 flume-log2.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a3 -c conf/ -f job_sample/interceptor/flume-log2.conf</span><br></pre></td></tr></table></figure>

<p>在 hadoop102，启动 exec-flume.conf</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/interceptor/exec-flume.conf</span><br></pre></td></tr></table></figure>

<p>启动成功后，确保没有报错后，在 hadoop102 上，往端口发送数据</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc localhost 44444</span><br></pre></td></tr></table></figure>

<h2 id="六、自定义-Source"><a href="#六、自定义-Source" class="headerlink" title="六、自定义 Source"></a>六、自定义 Source</h2><h3 id="自定义-Source-官方文档"><a href="#自定义-Source-官方文档" class="headerlink" title="自定义 Source 官方文档"></a>自定义 Source 官方文档</h3><p>Source 是负责接收数据到 Flume Agent 的组件。</p>
<p>Source 组件可以处理各种类型、各种格式的日志数据，官方已经提供了好多的现成的 Source 组件，例如：Exec Source、Spooling Directory Source、NetCat Source 还有 Taildir Source 等等。</p>
<p>当官方提供的 Source 组件无法满足我们实际需求的时候，我们就可以通过自定义 Source 来解决问题。</p>
<p>官方也提供了自定义 Source 的接口：<a target="_blank" rel="noopener" href="https://flume.apache.org/releases/content/1.11.0/FlumeDeveloperGuide.html#source">Flume 1.11.0 Developer Guide — Apache Flume</a></p>
<h3 id="案例需求-5"><a href="#案例需求-5" class="headerlink" title="案例需求"></a>案例需求</h3><p>通过自定义 Source 接收数据，并且给条数据前面添加个前缀，然后输出到控制台。</p>
<p>前缀可以在配置文件中灵活配置。</p>
<h3 id="需求分析-5"><a href="#需求分析-5" class="headerlink" title="需求分析"></a>需求分析</h3><ol>
<li>自定义 Source，使用随机数模拟产生数据；</li>
<li>将数据添加前缀，并封装成 Event 返回；</li>
<li>前缀与等待时间间隔都从配置文件中配置；</li>
</ol>
<h3 id="案例实现-5"><a href="#案例实现-5" class="headerlink" title="案例实现"></a>案例实现</h3><h4 id="ⅰ自定义-Source"><a href="#ⅰ自定义-Source" class="headerlink" title="ⅰ自定义 Source"></a>ⅰ自定义 Source</h4><ol>
<li>自定义 Source，并根据官方示例继承 AbstractSource，实现 Configurable 和 PollableSource</li>
</ol>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> org.example.source;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> org.apache.flume.Context;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.Event;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.EventDeliveryException;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.PollableSource;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.channel.ChannelProcessor;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.conf.Configurable;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.event.SimpleEvent;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.source.AbstractSource;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.nio.charset.Charset;</span><br><span class="line"><span class="keyword">import</span> java.util.HashMap;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@Description</span> TODO 自定义 Source</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * 需求：使用随机数模拟数据，并将数据添加前缀，数据前缀与时间间隔都通过配置文件配置</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * backoff 理解：就是在获取数据时，假如没有获取到数据，等待的一个时间</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">MySource</span> <span class="keyword">extends</span> <span class="title class_">AbstractSource</span> <span class="keyword">implements</span> <span class="title class_">Configurable</span>, PollableSource &#123;</span><br><span class="line"></span><br><span class="line">    <span class="comment">//前缀</span></span><br><span class="line">    <span class="keyword">private</span> String prefix;</span><br><span class="line">    <span class="comment">//程序暂停时间间隔</span></span><br><span class="line">    <span class="keyword">private</span> Long interval;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 初始化方法，可以通过context获取配置信息</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> context</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">configure</span><span class="params">(Context context)</span> &#123;</span><br><span class="line">        <span class="comment">//从配置文件中获取前缀和程度执行时间</span></span><br><span class="line">        prefix = context.getString(<span class="string">&quot;prefix&quot;</span>, <span class="string">&quot;default_prefix&quot;</span>);</span><br><span class="line">        interval = context.getLong(<span class="string">&quot;interval&quot;</span>, <span class="number">1000L</span>);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理数据的方法，方法返回数据处理状态</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     * <span class="doctag">@throws</span> EventDeliveryException</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> Status <span class="title function_">process</span><span class="params">()</span> <span class="keyword">throws</span> EventDeliveryException &#123;</span><br><span class="line"></span><br><span class="line">        <span class="type">Status</span> <span class="variable">status</span> <span class="operator">=</span> <span class="literal">null</span>;</span><br><span class="line">        <span class="type">ChannelProcessor</span> <span class="variable">processor</span> <span class="operator">=</span> getChannelProcessor();</span><br><span class="line"></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            <span class="comment">//获取数据</span></span><br><span class="line">            <span class="type">Event</span> <span class="variable">event</span> <span class="operator">=</span> getSomeData();</span><br><span class="line">            <span class="comment">//将Event交给 ChannelProcessor</span></span><br><span class="line">            processor.processEvent(event);</span><br><span class="line">            <span class="comment">//返回“准备”状态</span></span><br><span class="line">            status = Status.READY;</span><br><span class="line"></span><br><span class="line">            Thread.sleep(interval);</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Throwable t) &#123;</span><br><span class="line">            <span class="comment">//返回“错误”状态</span></span><br><span class="line">            status = Status.BACKOFF;</span><br><span class="line">            <span class="keyword">if</span> (t <span class="keyword">instanceof</span>  Error) &#123;</span><br><span class="line">                <span class="keyword">throw</span>  (Error) t;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> status;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * backoff 步长</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="type">long</span> <span class="title function_">getBackOffSleepIncrement</span><span class="params">()</span> &#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="number">1000</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * backoff 最长时间</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="type">long</span> <span class="title function_">getMaxBackOffSleepInterval</span><span class="params">()</span> &#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="number">10000</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 获取数据，并将数据包装成Event</span></span><br><span class="line"><span class="comment">     * </span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> Event <span class="title function_">getSomeData</span><span class="params">()</span> &#123;</span><br><span class="line">        <span class="comment">/**</span></span><br><span class="line"><span class="comment">         * 可以使用流的方式读入需要的数据，然后进行封装</span></span><br><span class="line"><span class="comment">         */</span></span><br><span class="line">        </span><br><span class="line">        <span class="comment">//这里使用随机数模拟数据</span></span><br><span class="line">        <span class="type">String</span> <span class="variable">data</span> <span class="operator">=</span> Double.toString(Math.random());</span><br><span class="line">        <span class="comment">//将数据封装</span></span><br><span class="line">        <span class="type">SimpleEvent</span> <span class="variable">event</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">SimpleEvent</span>();</span><br><span class="line">        event.setHeaders(<span class="keyword">new</span> <span class="title class_">HashMap</span>&lt;String, String&gt;());</span><br><span class="line">        event.setBody((prefix + data).getBytes(Charset.defaultCharset()));</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> event;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>将完成后的代码打包，并且上传到 <code>/opt/module/flume-1.11.0/lib/</code> 目录下</li>
</ol>
<h4 id="ⅱ编写Job配置文件-1"><a href="#ⅱ编写Job配置文件-1" class="headerlink" title="ⅱ编写Job配置文件"></a>ⅱ编写Job配置文件</h4><p>配置文件放到 <code>/opt/module/flume-1.11.0/job-sample/source/</code> 目录下。</p>
<p>创建 defsource-log.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">org.example.source.MySource</span></span><br><span class="line"><span class="attr">a1.sources.r1.prefix</span> = <span class="string">test111-</span></span><br><span class="line"><span class="attr">a1.sources.r1.interval</span> = <span class="string">500</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">logger</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<h3 id="启动测试-5"><a href="#启动测试-5" class="headerlink" title="启动测试"></a>启动测试</h3><p>将 defsource-log.conf 启动</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/source/defsource-log.conf</span><br></pre></td></tr></table></figure>

<p>查看日志，看产生的数据是否可以添加后缀</p>
<p>修改时间间隔，看是否有效</p>
<h2 id="七、自定义-Sink"><a href="#七、自定义-Sink" class="headerlink" title="七、自定义 Sink"></a>七、自定义 Sink</h2><h3 id="自定义-Sink-官方文档"><a href="#自定义-Sink-官方文档" class="headerlink" title="自定义 Sink 官方文档"></a>自定义 Sink 官方文档</h3><p>自定义 Sink 与 Source 基本一致，官方提供的自定义 Sink 的接口：<a target="_blank" rel="noopener" href="https://flume.apache.org/releases/content/1.11.0/FlumeDeveloperGuide.html#sink">Flume 1.11.0 Developer Guide — Apache Flume</a></p>
<h3 id="案例需求-6"><a href="#案例需求-6" class="headerlink" title="案例需求"></a>案例需求</h3><p>使用 Exec Source 监听某个端口，将接收到的数据传给 Sink，由自定义 Sink 实现添加后缀，并将数据打印到控制台。</p>
<h3 id="需求分析-6"><a href="#需求分析-6" class="headerlink" title="需求分析"></a>需求分析</h3><h3 id="案例实现-6"><a href="#案例实现-6" class="headerlink" title="案例实现"></a>案例实现</h3><h4 id="ⅰ自定义-Sink"><a href="#ⅰ自定义-Sink" class="headerlink" title="ⅰ自定义 Sink"></a>ⅰ自定义 Sink</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> org.example.sink;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> org.apache.flume.*;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.conf.Configurable;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.sink.AbstractSink;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.Logger;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.LoggerFactory;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@Description</span> TODO 自定义 Sink</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * 需求：将数据添加后缀，并打印到控制台</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">MySink</span> <span class="keyword">extends</span> <span class="title class_">AbstractSink</span> <span class="keyword">implements</span> <span class="title class_">Configurable</span> &#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="type">Logger</span> <span class="variable">logger</span> <span class="operator">=</span> LoggerFactory.getLogger(MySink.class);</span><br><span class="line"></span><br><span class="line">    <span class="comment">//后缀</span></span><br><span class="line">    <span class="keyword">private</span> String suffix;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 初始化方法，可以通过context获取配置信息</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> context</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">configure</span><span class="params">(Context context)</span> &#123;</span><br><span class="line">        suffix = context.getString(<span class="string">&quot;suffix&quot;</span>, <span class="string">&quot;def_suffix&quot;</span>);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理数据的方法</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     * <span class="doctag">@throws</span> EventDeliveryException</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> Status <span class="title function_">process</span><span class="params">()</span> <span class="keyword">throws</span> EventDeliveryException &#123;</span><br><span class="line">        <span class="type">Status</span> <span class="variable">status</span> <span class="operator">=</span> <span class="literal">null</span>;</span><br><span class="line"></span><br><span class="line">        <span class="comment">//获取Channel</span></span><br><span class="line">        <span class="type">Channel</span> <span class="variable">channel</span> <span class="operator">=</span> getChannel();</span><br><span class="line">        <span class="comment">//通过Channel获取事务</span></span><br><span class="line">        <span class="type">Transaction</span> <span class="variable">transaction</span> <span class="operator">=</span> channel.getTransaction();</span><br><span class="line">        <span class="comment">//开启事务</span></span><br><span class="line">        transaction.begin();</span><br><span class="line"></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            Event event;</span><br><span class="line"></span><br><span class="line">            <span class="comment">//从Channel中take数据</span></span><br><span class="line">            <span class="keyword">while</span> ((event = channel.take()) == <span class="literal">null</span>) &#123;</span><br><span class="line">                Thread.sleep(<span class="number">1000L</span>);</span><br><span class="line">            &#125;</span><br><span class="line">            <span class="comment">//将Event输出</span></span><br><span class="line">            write(event);</span><br><span class="line">            <span class="comment">//提交事务</span></span><br><span class="line">            transaction.commit();</span><br><span class="line">            status = Status.READY;</span><br><span class="line"></span><br><span class="line">        &#125; <span class="keyword">catch</span> (Throwable t) &#123;</span><br><span class="line">            <span class="comment">//报错，事务回滚</span></span><br><span class="line">            transaction.rollback();</span><br><span class="line">            status = Status.BACKOFF;</span><br><span class="line">            <span class="keyword">if</span> (t <span class="keyword">instanceof</span> Error) &#123;</span><br><span class="line">                <span class="keyword">throw</span>  (Error) t;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">            transaction.close();</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> status;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理事件的方法，给事件的主体添加后缀，并打印到控制台</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> event</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">void</span> <span class="title function_">write</span><span class="params">(Event event)</span> &#123;</span><br><span class="line">        <span class="type">String</span> <span class="variable">body</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">String</span>(event.getBody());</span><br><span class="line">        logger.info(body + suffix);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h4 id="ⅱ编写Job配置文件-2"><a href="#ⅱ编写Job配置文件-2" class="headerlink" title="ⅱ编写Job配置文件"></a>ⅱ编写Job配置文件</h4><p>配置文件放到 <code>/opt/module/flume-1.11.0/job-sample/sink/</code> 目录下。</p>
<p>创建 netcat-defsink.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">netcat</span></span><br><span class="line"><span class="attr">a1.sources.r1.bind</span> = <span class="string">localhost</span></span><br><span class="line"><span class="attr">a1.sources.r1.port</span> = <span class="string">44444</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">org.example.sink.MySink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.suffix</span> = <span class="string">-plus</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<h3 id="启动测试-6"><a href="#启动测试-6" class="headerlink" title="启动测试"></a>启动测试</h3><p>将 netcat-defsink.conf 启动</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent -n a1 -c conf/ -f job_sample/sink/netcat-defsink.conf</span><br></pre></td></tr></table></figure>

<p>再使用命令往端口发送数据</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc localhost 44444</span><br></pre></td></tr></table></figure>

<p>看数据是否添加了后缀</p>
<h3 id="需求扩展"><a href="#需求扩展" class="headerlink" title="需求扩展"></a>需求扩展</h3><h4 id="ⅰ扩展需求"><a href="#ⅰ扩展需求" class="headerlink" title="ⅰ扩展需求"></a>ⅰ扩展需求</h4><p>将自定义 Sink 修改为批量输出</p>
<h4 id="ⅱ编写自定义Sink"><a href="#ⅱ编写自定义Sink" class="headerlink" title="ⅱ编写自定义Sink"></a>ⅱ编写自定义Sink</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> org.example.sink;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> org.apache.flume.*;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.conf.Configurable;</span><br><span class="line"><span class="keyword">import</span> org.apache.flume.sink.AbstractSink;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.Logger;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.LoggerFactory;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util.ArrayList;</span><br><span class="line"><span class="keyword">import</span> java.util.List;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@Description</span>  补充：批量输出</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">class</span> <span class="title class_">MySinkBatch</span> <span class="keyword">extends</span> <span class="title class_">AbstractSink</span> <span class="keyword">implements</span> <span class="title class_">Configurable</span> &#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="type">Logger</span> <span class="variable">logger</span> <span class="operator">=</span> LoggerFactory.getLogger(MySinkBatch.class);</span><br><span class="line"></span><br><span class="line">    <span class="comment">//后缀</span></span><br><span class="line">    <span class="keyword">private</span> String suffix;</span><br><span class="line">    <span class="comment">//批次大小</span></span><br><span class="line">    <span class="keyword">private</span> <span class="type">int</span> batch;</span><br><span class="line">    <span class="keyword">private</span> List&lt;Event&gt; events = <span class="keyword">new</span> <span class="title class_">ArrayList</span>&lt;&gt;();</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 初始化方法，可以通过context获取配置信息</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> context</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">void</span> <span class="title function_">configure</span><span class="params">(Context context)</span> &#123;</span><br><span class="line">        suffix = context.getString(<span class="string">&quot;suffix&quot;</span>, <span class="string">&quot;def_suffix&quot;</span>);</span><br><span class="line">        batch = context.getInteger(<span class="string">&quot;batch&quot;</span>, <span class="number">10</span>);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理数据的方法</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span></span></span><br><span class="line"><span class="comment">     * <span class="doctag">@throws</span> EventDeliveryException</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> Status <span class="title function_">process</span><span class="params">()</span> <span class="keyword">throws</span> EventDeliveryException &#123;</span><br><span class="line">        <span class="type">Status</span> <span class="variable">status</span> <span class="operator">=</span> <span class="literal">null</span>;</span><br><span class="line"></span><br><span class="line">        <span class="comment">//清空Event集合</span></span><br><span class="line">        events.clear();</span><br><span class="line">        <span class="comment">//获取Channel</span></span><br><span class="line">        <span class="type">Channel</span> <span class="variable">channel</span> <span class="operator">=</span> getChannel();</span><br><span class="line">        <span class="comment">//通过Channel获取事务</span></span><br><span class="line">        <span class="type">Transaction</span> <span class="variable">transaction</span> <span class="operator">=</span> channel.getTransaction();</span><br><span class="line">        <span class="comment">//开启事务</span></span><br><span class="line">        transaction.begin();</span><br><span class="line"></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            <span class="comment">//循环batch，获取Event，然后添加到集合中</span></span><br><span class="line">            checkout:</span><br><span class="line">            <span class="keyword">for</span> (<span class="type">int</span> <span class="variable">i</span> <span class="operator">=</span> <span class="number">0</span>; i &lt; batch; i++) &#123;</span><br><span class="line">                Event event;</span><br><span class="line">                <span class="type">int</span> <span class="variable">count</span> <span class="operator">=</span> <span class="number">0</span>;</span><br><span class="line"></span><br><span class="line">                <span class="keyword">while</span> ((event = channel.take()) == <span class="literal">null</span>) &#123;</span><br><span class="line">                    Thread.sleep(<span class="number">1000L</span>);</span><br><span class="line">                    count++;</span><br><span class="line"></span><br><span class="line">                    <span class="keyword">if</span> (count &gt;= <span class="number">10</span>) &#123;</span><br><span class="line">                        <span class="keyword">break</span> checkout;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">                events.add(event);</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="comment">//处理Event</span></span><br><span class="line">            write(events);</span><br><span class="line">            <span class="comment">//提交事务</span></span><br><span class="line">            transaction.commit();</span><br><span class="line">            status = Status.READY;</span><br><span class="line"></span><br><span class="line">        &#125; <span class="keyword">catch</span> (Throwable t) &#123;</span><br><span class="line">            <span class="comment">//事务回滚</span></span><br><span class="line">            transaction.rollback();</span><br><span class="line">            status = Status.BACKOFF;</span><br><span class="line">            <span class="keyword">if</span> (t <span class="keyword">instanceof</span> Error) &#123;</span><br><span class="line">                <span class="keyword">throw</span> (Error) t;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">            transaction.close();</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> status;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理一批Event方法</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> events</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">void</span> <span class="title function_">write</span><span class="params">(List&lt;Event&gt; events)</span> &#123;</span><br><span class="line">        events.forEach(<span class="built_in">this</span>::write);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 处理Event方法</span></span><br><span class="line"><span class="comment">     * 1、获取Event的数据，添加后缀</span></span><br><span class="line"><span class="comment">     * 2、打印控制台</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> event</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">void</span> <span class="title function_">write</span><span class="params">(Event event)</span> &#123;</span><br><span class="line">        <span class="type">String</span> <span class="variable">body</span> <span class="operator">=</span> <span class="keyword">new</span> <span class="title class_">String</span>(event.getBody());</span><br><span class="line">        logger.info(body + suffix);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h4 id="ⅲ编写Job配置文件"><a href="#ⅲ编写Job配置文件" class="headerlink" title="ⅲ编写Job配置文件"></a>ⅲ编写Job配置文件</h4><p>配置文件放到 <code>/opt/module/flume-1.11.0/job-sample/sink/</code> 目录下。</p>
<p>创建 netcat-defsink_batch.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment"># Name the components on this agent</span></span><br><span class="line"><span class="attr">a1.sources</span> = <span class="string">r1</span></span><br><span class="line"><span class="attr">a1.sinks</span> = <span class="string">k1</span></span><br><span class="line"><span class="attr">a1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe/configure the source</span></span><br><span class="line"><span class="attr">a1.sources.r1.type</span> = <span class="string">netcat</span></span><br><span class="line"><span class="attr">a1.sources.r1.bind</span> = <span class="string">localhost</span></span><br><span class="line"><span class="attr">a1.sources.r1.port</span> = <span class="string">44444</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Use a channel which buffers events in memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.type</span> = <span class="string">memory</span></span><br><span class="line"><span class="attr">a1.channels.c1.capacity</span> = <span class="string">1000</span></span><br><span class="line"><span class="attr">a1.channels.c1.transactionCapacity</span> = <span class="string">100</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Describe the sink</span></span><br><span class="line"><span class="attr">a1.sinks.k1.type</span> = <span class="string">org.example.sink.MySinkBatch</span></span><br><span class="line"><span class="attr">a1.sinks.k1.suffix</span> = <span class="string">-plus</span></span><br><span class="line"><span class="attr">a1.sinks.k1.batch</span> = <span class="string">10</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment"># Bind the source and sink to the channel</span></span><br><span class="line"><span class="attr">a1.sources.r1.channels</span> = <span class="string">c1</span></span><br><span class="line"><span class="attr">a1.sinks.k1.channel</span> = <span class="string">c1</span></span><br></pre></td></tr></table></figure>

<h1 id="Flume-数据流监控-Ganglia"><a href="#Flume-数据流监控-Ganglia" class="headerlink" title="Flume 数据流监控 Ganglia"></a>Flume 数据流监控 Ganglia</h1><h2 id="一、Ganglia-安装"><a href="#一、Ganglia-安装" class="headerlink" title="一、Ganglia 安装"></a>一、Ganglia 安装</h2><h3 id="Ganglia-安装"><a href="#Ganglia-安装" class="headerlink" title="Ganglia 安装"></a>Ganglia 安装</h3><ol>
<li>组件部署规划</li>
</ol>
<table>
<thead>
<tr>
<th></th>
<th>web</th>
<th>gmetad</th>
<th>gmod</th>
</tr>
</thead>
<tbody><tr>
<td>hadoop102</td>
<td>✔</td>
<td>✔</td>
<td>✔</td>
</tr>
<tr>
<td>hadoop103</td>
<td></td>
<td></td>
<td>✔</td>
</tr>
<tr>
<td>hadoop104</td>
<td></td>
<td></td>
<td>✔</td>
</tr>
</tbody></table>
<ol start="2">
<li>分别在 hadoop102、hadoop103 和 hadoop104 机器执行安装 epel-release</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo yum -y install epel-release</span><br></pre></td></tr></table></figure>

<ol start="3">
<li>在 hadoop102 机器执行安装</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">sudo yum -y install ganglia-gmetad </span><br><span class="line">sudo yum -y install ganglia-web</span><br><span class="line">sudo yum -y install ganglia-gmond</span><br></pre></td></tr></table></figure>

<ol start="4">
<li>在 hadoop103 和 hadoop104 机器执行安装</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo yum -y install ganglia-gmond</span><br></pre></td></tr></table></figure>

<h3 id="Ganglia-修改配置文件"><a href="#Ganglia-修改配置文件" class="headerlink" title="Ganglia 修改配置文件"></a>Ganglia 修改配置文件</h3><ol>
<li>在 hadoop102 修改 ganglia.conf 配置文件</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo vim /etc/httpd/conf.d/ganglia.conf</span><br></pre></td></tr></table></figure>

<p>修改以下内容：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">#</span></span><br><span class="line"><span class="comment"># Ganglia monitoring system php web frontend</span></span><br><span class="line"><span class="comment">#</span></span><br><span class="line"></span><br><span class="line">Alias /ganglia /usr/share/ganglia</span><br><span class="line"></span><br><span class="line">&lt;Location /ganglia&gt;</span><br><span class="line">  <span class="comment"># Require local</span></span><br><span class="line">  <span class="comment"># Require ip 10.1.2.3</span></span><br><span class="line">  Require ip 192.168.8.1</span><br><span class="line">  <span class="comment"># Require host example.org</span></span><br><span class="line">&lt;/Location&gt;</span><br></pre></td></tr></table></figure>

<blockquote>
<p>注意：<br>这里配置的 IP 是虚拟机网上IP。<br>在 Windows 上，使用 ipconfig 命令查看，配置 VMware Network Adapter VMnet8 网卡的 IP。</p>
</blockquote>
<ol start="2">
<li>在 hadoop102 修改 gmetad.conf 配置文件</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo vim /etc/ganglia/gmetad.conf</span><br></pre></td></tr></table></figure>

<p>修改以下内容：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">data_source <span class="string">&quot;my cluster&quot;</span> hadoop102</span><br></pre></td></tr></table></figure>

<ol start="3">
<li>在所有机器上，修改 gmond.conf 配置文件</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo vim /etc/ganglia/gmond.conf</span><br></pre></td></tr></table></figure>

<p>修改以下内容：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line">cluster &#123;</span><br><span class="line">  name = <span class="string">&quot;my cluster&quot;</span>  <span class="comment">#修改为对应数据源名字</span></span><br><span class="line">  owner = <span class="string">&quot;unspecified&quot;</span></span><br><span class="line">  latlong = <span class="string">&quot;unspecified&quot;</span></span><br><span class="line">  url = <span class="string">&quot;unspecified&quot;</span></span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">udp_send_channel &#123;</span><br><span class="line">  <span class="comment">#bind_hostname = yes # Highly recommended, soon to be default.</span></span><br><span class="line">                       <span class="comment"># This option tells gmond to use a source address</span></span><br><span class="line">                       <span class="comment"># that resolves to the machine&#x27;s hostname.  Without</span></span><br><span class="line">                       <span class="comment"># this, the metrics may appear to come from any</span></span><br><span class="line">                       <span class="comment"># interface and the DNS names associated with</span></span><br><span class="line">                       <span class="comment"># those IPs will be used to create the RRDs.</span></span><br><span class="line">  <span class="comment">#mcast_join = 239.2.11.71</span></span><br><span class="line">  host = hadoop102  <span class="comment">#添加host主机名</span></span><br><span class="line">  port = 8649</span><br><span class="line">  ttl = 1</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">udp_recv_channel &#123;</span><br><span class="line">  <span class="comment">#mcast_join = 239.2.11.71</span></span><br><span class="line">  port = 8649</span><br><span class="line">  <span class="built_in">bind</span> = 0.0.0.0  <span class="comment">#绑定0.0.0.0，接收任意连接数据</span></span><br><span class="line">  retry_bind = <span class="literal">true</span></span><br><span class="line">  <span class="comment"># Size of the UDP buffer. If you are handling lots of metrics you really</span></span><br><span class="line">  <span class="comment"># should bump it up to e.g. 10MB or even higher.</span></span><br><span class="line">  <span class="comment"># buffer = 10485760</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<ol start="3">
<li>在 hadoop102 修改 confg 系统配置文件</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo vim /etc/selinux/config</span><br></pre></td></tr></table></figure>

<p>修改以下内容：</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">SELINUX=disabled</span><br></pre></td></tr></table></figure>

<blockquote>
<p>注意：selinux 需要重启，修改才会生效，如果此时不想重启，可以修改让其生效：<code>sudo setenforce 0</code></p>
</blockquote>
<h2 id="二、启动-Ganglia"><a href="#二、启动-Ganglia" class="headerlink" title="二、启动 Ganglia"></a>二、启动 Ganglia</h2><ol>
<li>在 hadoop102、hadoop103 和 hadoop104 启动</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo systemctl start gmond</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>在 hadoop102 启动</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">sudo systemctl start httpd</span><br><span class="line">sudo systemctl start gmetad</span><br></pre></td></tr></table></figure>

<h2 id="三、访问-Ganglia-的-web界面"><a href="#三、访问-Ganglia-的-web界面" class="headerlink" title="三、访问 Ganglia 的 web界面"></a>三、访问 Ganglia 的 web界面</h2><p><a target="_blank" rel="noopener" href="http://192.168.8.102/ganglia">http://192.168.8.102/ganglia</a></p>
<p>如果访问上面链接时，页面返回：You don’t have permission to access &#x2F;ganglia on this server.</p>
<p>说明权限不足，需要修改 &#x2F;var&#x2F;lib&#x2F;ganglib 配置文件</p>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sudo <span class="built_in">chmod</span> -R 777 /var/lib/ganglis</span><br></pre></td></tr></table></figure>

<h2 id="四、操作-Flume-测试监控"><a href="#四、操作-Flume-测试监控" class="headerlink" title="四、操作 Flume 测试监控"></a>四、操作 Flume 测试监控</h2><h3 id="启动测试-7"><a href="#启动测试-7" class="headerlink" title="启动测试"></a>启动测试</h3><ol>
<li>启动 Flume 的 Job</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line">./bin/flume-ng agent \</span><br><span class="line">-n a1 \</span><br><span class="line">-c conf/ \</span><br><span class="line">-f job_sample/flume-netcat-logger.conf \</span><br><span class="line">-Dflume.monitoring.type=ganglia \</span><br><span class="line">-Dflume.monitoring.hosts=192.168.8.102:8649</span><br></pre></td></tr></table></figure>

<ol start="2">
<li>往端口发送数据</li>
</ol>
<figure class="highlight sh"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">nc localhost 44444</span><br></pre></td></tr></table></figure>

<h3 id="字段说明"><a href="#字段说明" class="headerlink" title="字段说明"></a>字段说明</h3><table>
<thead>
<tr>
<th>字段（图表名称）</th>
<th>字段含义</th>
</tr>
</thead>
<tbody><tr>
<td>EventPutAttemptCount</td>
<td>source尝试写入channel的事件总数量</td>
</tr>
<tr>
<td>EventPutSuccessCount</td>
<td>成功写入channel且提交的事件总数量</td>
</tr>
<tr>
<td>EventTakeAttemptCount</td>
<td>sink尝试从channel拉取事件的总数量。</td>
</tr>
<tr>
<td>EventTakeSuccessCount</td>
<td>sink成功读取的事件的总数量</td>
</tr>
<tr>
<td>StartTime</td>
<td>channel启动的时间（毫秒）</td>
</tr>
<tr>
<td>StopTime</td>
<td>channel停止的时间（毫秒）</td>
</tr>
<tr>
<td>ChannelSize</td>
<td>目前channel中事件的总数量</td>
</tr>
<tr>
<td>ChannelFillPercentage</td>
<td>channel占用百分比</td>
</tr>
<tr>
<td>ChannelCapacity</td>
<td>channel的容量</td>
</tr>
</tbody></table>
<h1 id="企业真实面试题"><a href="#企业真实面试题" class="headerlink" title="企业真实面试题"></a>企业真实面试题</h1><h2 id="一、如何实现-Flume-监控"><a href="#一、如何实现-Flume-监控" class="headerlink" title="一、如何实现 Flume 监控"></a>一、如何实现 Flume 监控</h2><p>使用第三方框架 Ganglia 进行实时监控 Flume</p>
<h2 id="二、Flume-的-Source，Channel，Sink-的作用？你们使用的-Source-是什么类型？"><a href="#二、Flume-的-Source，Channel，Sink-的作用？你们使用的-Source-是什么类型？" class="headerlink" title="二、Flume 的 Source，Channel，Sink 的作用？你们使用的 Source 是什么类型？"></a>二、Flume 的 Source，Channel，Sink 的作用？你们使用的 Source 是什么类型？</h2><p>Source：该组件是专门用来收集数据的，可以处理各种类型、各种格式的日志数据；</p>
<p>Channel：该组件是将采集到的数据进行缓存，可以存放在内存（Memory）和文件（File）中；</p>
<p>Sink：该组件是把数据发送到目的地的组件；</p>
<p>监控后台日志使用的是 Taildir Source</p>
<h2 id="三、Flume的Channel-Selectors"><a href="#三、Flume的Channel-Selectors" class="headerlink" title="三、Flume的Channel Selectors"></a>三、Flume的Channel Selectors</h2><p>作用：将数据发往不同的 Channel 中，从而进一步通过 Sink，到下游的系统中；</p>
<p>类型：Replicating Channel Selector（default） 和 Multiplexing Channel Selector</p>
<ul>
<li>Replicating：发往所有 Channel；</li>
<li>Multiplexing：发往指定 Channel；</li>
</ul>
<h2 id="四、Flume的事务机制"><a href="#四、Flume的事务机制" class="headerlink" title="四、Flume的事务机制"></a>四、Flume的事务机制</h2><p>put事务</p>
<p>take事务</p>
<h2 id="五、Flume采集数据会丢失吗？"><a href="#五、Flume采集数据会丢失吗？" class="headerlink" title="五、Flume采集数据会丢失吗？"></a>五、Flume采集数据会丢失吗？</h2><p>由于 Flume 内部架构原理，Flume 是不可能丢失数据的，其内部有完善的事务机制，Source 到 Channel 是事务性的，Channel 到 Sink 是事务性的，因此，这两个环节不会出现数据的丢失；</p>
<p>唯一可能丢失数据的情况，是使用 Memory Channel，agent 宕机导致数据丢失，或者 Channel 存储数据已满，导致 Source 不再写入，未写入数据会丢失；</p>
<p>Flume 采集数据不会丢失数据，但是有可能会造成数据重复；</p>
<ul>
<li>例：当 Sink 发出数据后，没有收到响应，Sink 会再次发送数据，这时可能会造成数据重复；</li>
</ul>
</article><div class="post-copyright"><div class="post-copyright__author"><span class="post-copyright-meta">文章作者: </span><span class="post-copyright-info"><a href="mailto:undefined">Oct25-X</a></span></div><div class="post-copyright__type"><span class="post-copyright-meta">文章链接: </span><span class="post-copyright-info"><a href="https://oct25-x.atomgit.net/oct25-xxxxx/2024/01/03/42%E6%95%B0%E6%8D%AE%E9%87%87%E9%9B%86%E7%B3%BB%E7%BB%9F-Flume/">https://oct25-x.atomgit.net/oct25-xxxxx/2024/01/03/42%E6%95%B0%E6%8D%AE%E9%87%87%E9%9B%86%E7%B3%BB%E7%BB%9F-Flume/</a></span></div><div class="post-copyright__notice"><span class="post-copyright-meta">版权声明: </span><span class="post-copyright-info">本博客所有文章除特别声明外，均采用 <a href="https://creativecommons.org/licenses/by-nc-sa/4.0/" target="_blank">CC BY-NC-SA 4.0</a> 许可协议。转载请注明来自 <a href="https://oct25-x.atomgit.net/oct25-xxxxx" target="_blank">Oct25-X</a>！</span></div></div><div class="tag_share"><div class="post-meta__tag-list"><a class="post-meta__tags" href="/oct25-xxxxx/tags/Flume/">Flume</a></div><div class="post_share"><div class="social-share" data-image="/oct25-xxxxx/img/cover_img/ctt42.png" data-sites="facebook,twitter,wechat,weibo,qq"></div><link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/social-share.js/dist/css/share.min.css" media="print" onload="this.media='all'"><script src="https://cdn.jsdelivr.net/npm/social-share.js/dist/js/social-share.min.js" defer></script></div></div><nav class="pagination-post" id="pagination"><div class="prev-post pull-left"><a href="/oct25-xxxxx/2024/01/07/43%E6%95%B0%E6%8D%AE%E9%87%87%E9%9B%86%E7%B3%BB%E7%BB%9FFlume%E5%AE%9E%E6%88%98/"><img class="prev-cover" src="/oct25-xxxxx/img/cover_img/ctt43.png" onerror="onerror=null;src='/oct25-xxxxx/img/404.jpg'" alt="cover of previous post"><div class="pagination-info"><div class="label">上一篇</div><div class="prev_info">数据采集系统 Flume 实战</div></div></a></div><div class="next-post pull-right"><a href="/oct25-xxxxx/2023/12/31/41ubuntu-%E5%AE%89%E8%A3%85%E9%83%A8%E7%BD%B2/"><img class="next-cover" src="/oct25-xxxxx/img/cover_img/ctt41.png" onerror="onerror=null;src='/oct25-xxxxx/img/404.jpg'" alt="cover of next post"><div class="pagination-info"><div class="label">下一篇</div><div class="next_info">ubuntu 安装 Java/MySQL/Redis</div></div></a></div></nav><div class="relatedPosts"><div class="headline"><i class="fas fa-thumbs-up fa-fw"></i><span>相关推荐</span></div><div class="relatedPosts-list"><div><a href="/oct25-xxxxx/2024/01/07/43%E6%95%B0%E6%8D%AE%E9%87%87%E9%9B%86%E7%B3%BB%E7%BB%9FFlume%E5%AE%9E%E6%88%98/" title="数据采集系统 Flume 实战"><img class="cover" src="/oct25-xxxxx/img/cover_img/ctt43.png" alt="cover"><div class="content is-center"><div class="date"><i class="far fa-calendar-alt fa-fw"></i> 2024-01-07</div><div class="title">数据采集系统 Flume 实战</div></div></a></div></div></div></div><div class="aside-content" id="aside-content"><div class="card-widget card-info"><div class="is-center"><div class="avatar-img"><img src="/oct25-xxxxx/img/avatar.jpg" onerror="this.onerror=null;this.src='/oct25-xxxxx/img/friend_404.gif'" alt="avatar"/></div><div class="author-info__name">Oct25-X</div><div class="author-info__description"></div></div><div class="card-info-data is-center"><div class="card-info-data-item"><a href="/oct25-xxxxx/archives/"><div class="headline">文章</div><div class="length-num">52</div></a></div><div class="card-info-data-item"><a href="/oct25-xxxxx/tags/"><div class="headline">标签</div><div class="length-num">17</div></a></div><div class="card-info-data-item"><a href="/oct25-xxxxx/categories/"><div class="headline">分类</div><div class="length-num">8</div></a></div></div><a id="card-info-btn" target="_blank" rel="noopener" href="https://github.com/xxxxxx"><i class="fab fa-github"></i><span>Follow Me</span></a></div><div class="sticky_layout"><div class="card-widget" id="card-toc"><div class="item-headline"><i class="fas fa-stream"></i><span>目录</span><span class="toc-percentage"></span></div><div class="toc-content"><ol class="toc"><li class="toc-item toc-level-1"><a class="toc-link" href="#Flume-%E6%A6%82%E8%BF%B0"><span class="toc-number">1.</span> <span class="toc-text">Flume 概述</span></a><ol class="toc-child"><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%80%E3%80%81Flume-%E5%AE%9A%E4%B9%89"><span class="toc-number">1.1.</span> <span class="toc-text">一、Flume 定义</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%8C%E3%80%81Flume-%E5%9F%BA%E7%A1%80%E6%9E%B6%E6%9E%84"><span class="toc-number">1.2.</span> <span class="toc-text">二、Flume 基础架构</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#Agent"><span class="toc-number">1.2.1.</span> <span class="toc-text">Agent</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#Source"><span class="toc-number">1.2.2.</span> <span class="toc-text">Source</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#Channel"><span class="toc-number">1.2.3.</span> <span class="toc-text">Channel</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#Sink"><span class="toc-number">1.2.4.</span> <span class="toc-text">Sink</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#Event"><span class="toc-number">1.2.5.</span> <span class="toc-text">Event</span></a></li></ol></li></ol></li><li class="toc-item toc-level-1"><a class="toc-link" href="#Flume-%E5%BF%AB%E9%80%9F%E5%85%A5%E9%97%A8"><span class="toc-number">2.</span> <span class="toc-text">Flume 快速入门</span></a><ol class="toc-child"><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%80%E3%80%81%E5%AE%98%E7%BD%91%E5%9C%B0%E5%9D%80"><span class="toc-number">2.1.</span> <span class="toc-text">一、官网地址</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%8C%E3%80%81Flume-%E5%AE%89%E8%A3%85%E9%83%A8%E7%BD%B2"><span class="toc-number">2.2.</span> <span class="toc-text">二、Flume 安装部署</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E4%B8%8A%E4%BC%A0-tar-%E5%8C%85%EF%BC%8C%E5%B9%B6%E8%A7%A3%E5%8E%8B"><span class="toc-number">2.2.1.</span> <span class="toc-text">上传 tar 包，并解压</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%88%A0%E9%99%A4%E4%BE%9D%E8%B5%96%E5%8C%85"><span class="toc-number">2.2.2.</span> <span class="toc-text">删除依赖包</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E4%BF%AE%E6%94%B9%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6"><span class="toc-number">2.2.3.</span> <span class="toc-text">修改配置文件</span></a></li></ol></li></ol></li><li class="toc-item toc-level-1"><a class="toc-link" href="#Flume-%E5%85%A5%E9%97%A8%E6%A1%88%E4%BE%8B"><span class="toc-number">3.</span> <span class="toc-text">Flume 入门案例</span></a><ol class="toc-child"><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%80%E3%80%81Flume-%E5%AE%98%E6%96%B9%E6%A1%88%E4%BE%8B"><span class="toc-number">3.1.</span> <span class="toc-text">一、Flume 官方案例</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82"><span class="toc-number">3.1.1.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90"><span class="toc-number">3.1.2.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%87%86%E5%A4%87%E5%B7%A5%E4%BD%9C"><span class="toc-number">3.1.3.</span> <span class="toc-text">准备工作</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0"><span class="toc-number">3.1.4.</span> <span class="toc-text">案例实现</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95"><span class="toc-number">3.1.5.</span> <span class="toc-text">启动测试</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%97%AE%E9%A2%98"><span class="toc-number">3.1.6.</span> <span class="toc-text">问题</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%8C%E3%80%81%E5%AE%9E%E6%97%B6%E7%9B%91%E6%8E%A7%E6%9C%AC%E5%9C%B0%E6%96%87%E4%BB%B6%E5%88%B0-HDFS"><span class="toc-number">3.2.</span> <span class="toc-text">二、实时监控本地文件到 HDFS</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-1"><span class="toc-number">3.2.1.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90-1"><span class="toc-number">3.2.2.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%87%86%E5%A4%87%E5%B7%A5%E4%BD%9C-1"><span class="toc-number">3.2.3.</span> <span class="toc-text">准备工作</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-1"><span class="toc-number">3.2.4.</span> <span class="toc-text">案例实现</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-1"><span class="toc-number">3.2.5.</span> <span class="toc-text">启动测试</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%89%E3%80%81%E5%AE%9E%E6%97%B6%E7%9B%91%E6%8E%A7%E6%8C%87%E5%AE%9A%E7%9B%AE%E5%BD%95%E4%B8%8B%E5%A4%9A%E4%B8%AA%E6%96%87%E4%BB%B6"><span class="toc-number">3.3.</span> <span class="toc-text">三、实时监控指定目录下多个文件</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-2"><span class="toc-number">3.3.1.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90-2"><span class="toc-number">3.3.2.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-2"><span class="toc-number">3.3.3.</span> <span class="toc-text">案例实现</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-2"><span class="toc-number">3.3.4.</span> <span class="toc-text">启动测试</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E5%9B%9B%E3%80%81%E5%AE%9E%E6%97%B6%E7%9B%91%E6%8E%A7%E7%9B%AE%E5%BD%95%E4%B8%8B%E7%9A%84%E5%A4%9A%E4%B8%AA%E8%BF%BD%E5%8A%A0%E6%96%87%E4%BB%B6"><span class="toc-number">3.4.</span> <span class="toc-text">四、实时监控目录下的多个追加文件</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-3"><span class="toc-number">3.4.1.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90-3"><span class="toc-number">3.4.2.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-3"><span class="toc-number">3.4.3.</span> <span class="toc-text">案例实现</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-3"><span class="toc-number">3.4.4.</span> <span class="toc-text">启动测试</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%97%AE%E9%A2%98-1"><span class="toc-number">3.4.5.</span> <span class="toc-text">问题</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E9%97%AE%E9%A2%98%E8%AF%B4%E6%98%8E"><span class="toc-number">3.4.5.1.</span> <span class="toc-text">ⅰ问题说明</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E9%97%AE%E9%A2%98%E5%A4%8D%E7%8E%B0"><span class="toc-number">3.4.5.2.</span> <span class="toc-text">ⅱ问题复现</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B2-%E4%BF%AE%E6%94%B9%E6%BA%90%E7%A0%81%E8%A7%A3%E5%86%B3"><span class="toc-number">3.4.5.3.</span> <span class="toc-text">ⅲ 修改源码解决</span></a></li></ol></li></ol></li></ol></li><li class="toc-item toc-level-1"><a class="toc-link" href="#Flume-%E8%BF%9B%E9%98%B6"><span class="toc-number">4.</span> <span class="toc-text">Flume 进阶</span></a><ol class="toc-child"><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%80%E3%80%81Flume-%E4%BA%8B%E5%8A%A1"><span class="toc-number">4.1.</span> <span class="toc-text">一、Flume 事务</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%8C%E3%80%81Flume-Agent-%E5%86%85%E9%83%A8%E5%8E%9F%E7%90%86"><span class="toc-number">4.2.</span> <span class="toc-text">二、Flume Agent 内部原理</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%89%E3%80%81Flume-%E6%8B%93%E6%89%91%E7%BB%93%E6%9E%84"><span class="toc-number">4.3.</span> <span class="toc-text">三、Flume 拓扑结构</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E7%AE%80%E5%8D%95%E4%B8%B2%E8%81%94"><span class="toc-number">4.3.1.</span> <span class="toc-text">简单串联</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%A4%8D%E5%88%B6%E5%92%8C%E5%A4%9A%E8%B7%AF%E5%A4%8D%E7%94%A8"><span class="toc-number">4.3.2.</span> <span class="toc-text">复制和多路复用</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E8%B4%9F%E8%BD%BD%E5%9D%87%E8%A1%A1%E5%92%8C%E6%95%85%E9%9A%9C%E8%BD%AC%E7%A7%BB"><span class="toc-number">4.3.3.</span> <span class="toc-text">负载均衡和故障转移</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E5%9B%9B%E3%80%81Flume-%E4%BC%81%E4%B8%9A%E5%BC%80%E5%8F%91%E6%A1%88%E4%BE%8B"><span class="toc-number">4.4.</span> <span class="toc-text">四、Flume 企业开发案例</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%A4%8D%E5%88%B6%E5%92%8C%E5%A4%9A%E8%B7%AF%E5%A4%8D%E7%94%A8-1"><span class="toc-number">4.4.1.</span> <span class="toc-text">复制和多路复用</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82"><span class="toc-number">4.4.1.1.</span> <span class="toc-text">ⅰ案例需求</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0"><span class="toc-number">4.4.1.2.</span> <span class="toc-text">ⅱ案例实现</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B2%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95"><span class="toc-number">4.4.1.3.</span> <span class="toc-text">ⅲ启动测试</span></a></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E8%B4%9F%E8%BD%BD%E5%9D%87%E8%A1%A1%E5%92%8C%E6%95%85%E9%9A%9C%E8%BD%AC%E7%A7%BB-1"><span class="toc-number">4.4.2.</span> <span class="toc-text">负载均衡和故障转移</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-1"><span class="toc-number">4.4.2.1.</span> <span class="toc-text">ⅰ案例需求</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-1"><span class="toc-number">4.4.2.2.</span> <span class="toc-text">ⅱ案例实现</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B2%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-1"><span class="toc-number">4.4.2.3.</span> <span class="toc-text">ⅲ启动测试</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B3%E8%B4%9F%E8%BD%BD%E5%9D%87%E8%A1%A1"><span class="toc-number">4.4.2.4.</span> <span class="toc-text">ⅳ负载均衡</span></a></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E8%81%9A%E5%90%88"><span class="toc-number">4.4.3.</span> <span class="toc-text">聚合</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-2"><span class="toc-number">4.4.3.1.</span> <span class="toc-text">ⅰ案例需求</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-2"><span class="toc-number">4.4.3.2.</span> <span class="toc-text">ⅱ案例实现</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B2%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-2"><span class="toc-number">4.4.3.3.</span> <span class="toc-text">ⅲ启动测试</span></a></li></ol></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%94%E3%80%81%E8%87%AA%E5%AE%9A%E4%B9%89%E6%8B%A6%E6%88%AA%E5%99%A8-Interceptor"><span class="toc-number">4.5.</span> <span class="toc-text">五、自定义拦截器 Interceptor</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%89%8D%E8%A8%80"><span class="toc-number">4.5.1.</span> <span class="toc-text">前言</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-4"><span class="toc-number">4.5.2.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90-4"><span class="toc-number">4.5.3.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-4"><span class="toc-number">4.5.4.</span> <span class="toc-text">案例实现</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E8%87%AA%E5%AE%9A%E4%B9%89-Interceptor"><span class="toc-number">4.5.4.1.</span> <span class="toc-text">ⅰ自定义 Interceptor</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E7%BC%96%E5%86%99Job%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6"><span class="toc-number">4.5.4.2.</span> <span class="toc-text">ⅱ编写Job配置文件</span></a></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-4"><span class="toc-number">4.5.5.</span> <span class="toc-text">启动测试</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E5%85%AD%E3%80%81%E8%87%AA%E5%AE%9A%E4%B9%89-Source"><span class="toc-number">4.6.</span> <span class="toc-text">六、自定义 Source</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E8%87%AA%E5%AE%9A%E4%B9%89-Source-%E5%AE%98%E6%96%B9%E6%96%87%E6%A1%A3"><span class="toc-number">4.6.1.</span> <span class="toc-text">自定义 Source 官方文档</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-5"><span class="toc-number">4.6.2.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90-5"><span class="toc-number">4.6.3.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-5"><span class="toc-number">4.6.4.</span> <span class="toc-text">案例实现</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E8%87%AA%E5%AE%9A%E4%B9%89-Source"><span class="toc-number">4.6.4.1.</span> <span class="toc-text">ⅰ自定义 Source</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E7%BC%96%E5%86%99Job%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6-1"><span class="toc-number">4.6.4.2.</span> <span class="toc-text">ⅱ编写Job配置文件</span></a></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-5"><span class="toc-number">4.6.5.</span> <span class="toc-text">启动测试</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%83%E3%80%81%E8%87%AA%E5%AE%9A%E4%B9%89-Sink"><span class="toc-number">4.7.</span> <span class="toc-text">七、自定义 Sink</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E8%87%AA%E5%AE%9A%E4%B9%89-Sink-%E5%AE%98%E6%96%B9%E6%96%87%E6%A1%A3"><span class="toc-number">4.7.1.</span> <span class="toc-text">自定义 Sink 官方文档</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E9%9C%80%E6%B1%82-6"><span class="toc-number">4.7.2.</span> <span class="toc-text">案例需求</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E5%88%86%E6%9E%90-6"><span class="toc-number">4.7.3.</span> <span class="toc-text">需求分析</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E6%A1%88%E4%BE%8B%E5%AE%9E%E7%8E%B0-6"><span class="toc-number">4.7.4.</span> <span class="toc-text">案例实现</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E8%87%AA%E5%AE%9A%E4%B9%89-Sink"><span class="toc-number">4.7.4.1.</span> <span class="toc-text">ⅰ自定义 Sink</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E7%BC%96%E5%86%99Job%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6-2"><span class="toc-number">4.7.4.2.</span> <span class="toc-text">ⅱ编写Job配置文件</span></a></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-6"><span class="toc-number">4.7.5.</span> <span class="toc-text">启动测试</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E9%9C%80%E6%B1%82%E6%89%A9%E5%B1%95"><span class="toc-number">4.7.6.</span> <span class="toc-text">需求扩展</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B0%E6%89%A9%E5%B1%95%E9%9C%80%E6%B1%82"><span class="toc-number">4.7.6.1.</span> <span class="toc-text">ⅰ扩展需求</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B1%E7%BC%96%E5%86%99%E8%87%AA%E5%AE%9A%E4%B9%89Sink"><span class="toc-number">4.7.6.2.</span> <span class="toc-text">ⅱ编写自定义Sink</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#%E2%85%B2%E7%BC%96%E5%86%99Job%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6"><span class="toc-number">4.7.6.3.</span> <span class="toc-text">ⅲ编写Job配置文件</span></a></li></ol></li></ol></li></ol></li><li class="toc-item toc-level-1"><a class="toc-link" href="#Flume-%E6%95%B0%E6%8D%AE%E6%B5%81%E7%9B%91%E6%8E%A7-Ganglia"><span class="toc-number">5.</span> <span class="toc-text">Flume 数据流监控 Ganglia</span></a><ol class="toc-child"><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%80%E3%80%81Ganglia-%E5%AE%89%E8%A3%85"><span class="toc-number">5.1.</span> <span class="toc-text">一、Ganglia 安装</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#Ganglia-%E5%AE%89%E8%A3%85"><span class="toc-number">5.1.1.</span> <span class="toc-text">Ganglia 安装</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#Ganglia-%E4%BF%AE%E6%94%B9%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6"><span class="toc-number">5.1.2.</span> <span class="toc-text">Ganglia 修改配置文件</span></a></li></ol></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%8C%E3%80%81%E5%90%AF%E5%8A%A8-Ganglia"><span class="toc-number">5.2.</span> <span class="toc-text">二、启动 Ganglia</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%89%E3%80%81%E8%AE%BF%E9%97%AE-Ganglia-%E7%9A%84-web%E7%95%8C%E9%9D%A2"><span class="toc-number">5.3.</span> <span class="toc-text">三、访问 Ganglia 的 web界面</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E5%9B%9B%E3%80%81%E6%93%8D%E4%BD%9C-Flume-%E6%B5%8B%E8%AF%95%E7%9B%91%E6%8E%A7"><span class="toc-number">5.4.</span> <span class="toc-text">四、操作 Flume 测试监控</span></a><ol class="toc-child"><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%90%AF%E5%8A%A8%E6%B5%8B%E8%AF%95-7"><span class="toc-number">5.4.1.</span> <span class="toc-text">启动测试</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#%E5%AD%97%E6%AE%B5%E8%AF%B4%E6%98%8E"><span class="toc-number">5.4.2.</span> <span class="toc-text">字段说明</span></a></li></ol></li></ol></li><li class="toc-item toc-level-1"><a class="toc-link" href="#%E4%BC%81%E4%B8%9A%E7%9C%9F%E5%AE%9E%E9%9D%A2%E8%AF%95%E9%A2%98"><span class="toc-number">6.</span> <span class="toc-text">企业真实面试题</span></a><ol class="toc-child"><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%80%E3%80%81%E5%A6%82%E4%BD%95%E5%AE%9E%E7%8E%B0-Flume-%E7%9B%91%E6%8E%A7"><span class="toc-number">6.1.</span> <span class="toc-text">一、如何实现 Flume 监控</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%8C%E3%80%81Flume-%E7%9A%84-Source%EF%BC%8CChannel%EF%BC%8CSink-%E7%9A%84%E4%BD%9C%E7%94%A8%EF%BC%9F%E4%BD%A0%E4%BB%AC%E4%BD%BF%E7%94%A8%E7%9A%84-Source-%E6%98%AF%E4%BB%80%E4%B9%88%E7%B1%BB%E5%9E%8B%EF%BC%9F"><span class="toc-number">6.2.</span> <span class="toc-text">二、Flume 的 Source，Channel，Sink 的作用？你们使用的 Source 是什么类型？</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%B8%89%E3%80%81Flume%E7%9A%84Channel-Selectors"><span class="toc-number">6.3.</span> <span class="toc-text">三、Flume的Channel Selectors</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E5%9B%9B%E3%80%81Flume%E7%9A%84%E4%BA%8B%E5%8A%A1%E6%9C%BA%E5%88%B6"><span class="toc-number">6.4.</span> <span class="toc-text">四、Flume的事务机制</span></a></li><li class="toc-item toc-level-2"><a class="toc-link" href="#%E4%BA%94%E3%80%81Flume%E9%87%87%E9%9B%86%E6%95%B0%E6%8D%AE%E4%BC%9A%E4%B8%A2%E5%A4%B1%E5%90%97%EF%BC%9F"><span class="toc-number">6.5.</span> <span class="toc-text">五、Flume采集数据会丢失吗？</span></a></li></ol></li></ol></div></div><div class="card-widget card-recent-post"><div class="item-headline"><i class="fas fa-history"></i><span>最新文章</span></div><div class="aside-list"><div class="aside-list-item"><a class="thumbnail" href="/oct25-xxxxx/2024/02/24/52%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala09_%E9%9A%90%E5%BC%8F%E8%BD%AC%E6%8D%A2%E3%80%81%E5%BC%82%E5%B8%B8%E5%92%8C%E6%B3%9B%E5%9E%8B/" title="大数据之多范式编程语言 Scala09_隐式转换、异常和泛型"><img src="/oct25-xxxxx/img/cover_img/ctt52.png" onerror="this.onerror=null;this.src='/oct25-xxxxx/img/404.jpg'" alt="大数据之多范式编程语言 Scala09_隐式转换、异常和泛型"/></a><div class="content"><a class="title" href="/oct25-xxxxx/2024/02/24/52%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala09_%E9%9A%90%E5%BC%8F%E8%BD%AC%E6%8D%A2%E3%80%81%E5%BC%82%E5%B8%B8%E5%92%8C%E6%B3%9B%E5%9E%8B/" title="大数据之多范式编程语言 Scala09_隐式转换、异常和泛型">大数据之多范式编程语言 Scala09_隐式转换、异常和泛型</a><time datetime="2024-02-24T04:55:34.000Z" title="发表于 2024-02-24 12:55:34">2024-02-24</time></div></div><div class="aside-list-item"><a class="thumbnail" href="/oct25-xxxxx/2024/02/24/51%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala08_%E6%A8%A1%E5%BC%8F%E5%8C%B9%E9%85%8D/" title="大数据之多范式编程语言 Scala08_模式匹配"><img src="/oct25-xxxxx/img/cover_img/ctt51.png" onerror="this.onerror=null;this.src='/oct25-xxxxx/img/404.jpg'" alt="大数据之多范式编程语言 Scala08_模式匹配"/></a><div class="content"><a class="title" href="/oct25-xxxxx/2024/02/24/51%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala08_%E6%A8%A1%E5%BC%8F%E5%8C%B9%E9%85%8D/" title="大数据之多范式编程语言 Scala08_模式匹配">大数据之多范式编程语言 Scala08_模式匹配</a><time datetime="2024-02-24T00:46:12.000Z" title="发表于 2024-02-24 08:46:12">2024-02-24</time></div></div><div class="aside-list-item"><a class="thumbnail" href="/oct25-xxxxx/2024/02/22/50%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala07_%E9%9B%86%E5%90%88/" title="大数据之多范式编程语言 Scala07_集合"><img src="/oct25-xxxxx/img/cover_img/ctt50.png" onerror="this.onerror=null;this.src='/oct25-xxxxx/img/404.jpg'" alt="大数据之多范式编程语言 Scala07_集合"/></a><div class="content"><a class="title" href="/oct25-xxxxx/2024/02/22/50%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala07_%E9%9B%86%E5%90%88/" title="大数据之多范式编程语言 Scala07_集合">大数据之多范式编程语言 Scala07_集合</a><time datetime="2024-02-22T08:16:10.000Z" title="发表于 2024-02-22 16:16:10">2024-02-22</time></div></div><div class="aside-list-item"><a class="thumbnail" href="/oct25-xxxxx/2024/02/21/49%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala06_%E9%9D%A2%E5%90%91%E5%AF%B9%E8%B1%A1/" title="大数据之多范式编程语言 Scala06_面向对象"><img src="/oct25-xxxxx/img/cover_img/ctt49.png" onerror="this.onerror=null;this.src='/oct25-xxxxx/img/404.jpg'" alt="大数据之多范式编程语言 Scala06_面向对象"/></a><div class="content"><a class="title" href="/oct25-xxxxx/2024/02/21/49%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala06_%E9%9D%A2%E5%90%91%E5%AF%B9%E8%B1%A1/" title="大数据之多范式编程语言 Scala06_面向对象">大数据之多范式编程语言 Scala06_面向对象</a><time datetime="2024-02-21T07:01:10.000Z" title="发表于 2024-02-21 15:01:10">2024-02-21</time></div></div><div class="aside-list-item"><a class="thumbnail" href="/oct25-xxxxx/2024/02/20/48%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala05_%E5%87%BD%E6%95%B0%E5%BC%8F%E7%BC%96%E7%A8%8B/" title="大数据之多范式编程语言 Scala05_函数式编程"><img src="/oct25-xxxxx/img/cover_img/ctt48.png" onerror="this.onerror=null;this.src='/oct25-xxxxx/img/404.jpg'" alt="大数据之多范式编程语言 Scala05_函数式编程"/></a><div class="content"><a class="title" href="/oct25-xxxxx/2024/02/20/48%E5%A4%A7%E6%95%B0%E6%8D%AE%E4%B9%8B%E5%A4%9A%E8%8C%83%E5%BC%8F%E7%BC%96%E7%A8%8B%E8%AF%AD%E8%A8%80Scala05_%E5%87%BD%E6%95%B0%E5%BC%8F%E7%BC%96%E7%A8%8B/" title="大数据之多范式编程语言 Scala05_函数式编程">大数据之多范式编程语言 Scala05_函数式编程</a><time datetime="2024-02-20T12:05:11.000Z" title="发表于 2024-02-20 20:05:11">2024-02-20</time></div></div></div></div></div></div></main><footer id="footer" style="background-image: url('/oct25-xxxxx/img/top_img/top_bg16.jpg')"><div id="footer-wrap"><div class="copyright">&copy;2020 - 2024 By Oct25-X</div><div class="framework-info"><span>框架 </span><a target="_blank" rel="noopener" href="https://hexo.io">Hexo</a><span class="footer-separator">|</span><span>主题 </span><a target="_blank" rel="noopener" href="https://github.com/jerryc127/hexo-theme-butterfly">Butterfly</a></div></div></footer></div><div id="rightside"><div id="rightside-config-hide"><button id="readmode" type="button" title="阅读模式"><i class="fas fa-book-open"></i></button><button id="translateLink" type="button" title="简繁转换">繁</button><button id="darkmode" type="button" title="浅色和深色模式转换"><i class="fas fa-adjust"></i></button><button id="hide-aside-btn" type="button" title="单栏和双栏切换"><i class="fas fa-arrows-alt-h"></i></button></div><div id="rightside-config-show"><button id="rightside_config" type="button" title="设置"><i class="fas fa-cog fa-spin"></i></button><button class="close" id="mobile-toc-button" type="button" title="目录"><i class="fas fa-list-ul"></i></button><button id="go-up" type="button" title="回到顶部"><i class="fas fa-arrow-up"></i></button></div></div><div id="local-search"><div class="search-dialog"><nav class="search-nav"><span class="search-dialog-title">本地搜索</span><span id="loading-status"></span><button class="search-close-button"><i class="fas fa-times"></i></button></nav><div class="is-center" id="loading-database"><i class="fas fa-spinner fa-pulse"></i><span>  数据库加载中</span></div><div class="search-wrap"><div id="local-search-input"><div class="local-search-box"><input class="local-search-box--input" placeholder="搜索文章" type="text"/></div></div><hr/><div id="local-search-results"></div></div></div><div id="search-mask"></div></div><div id="rightMenu"><div class="rightMenu-group rightMenu-small"><div class="rightMenu-item" id="menu-backward"><i class="fa-solid fa-arrow-left"></i></div><div class="rightMenu-item" id="menu-forward"><i class="fa-solid fa-arrow-right"></i></div><div class="rightMenu-item" id="menu-refresh"><i class="fa-solid fa-arrow-rotate-right"></i></div><div class="rightMenu-item" id="menu-home"><i class="fa-solid fa-house"></i></div></div><div class="rightMenu-group rightMenu-line rightMenuOther"><a class="rightMenu-item menu-link" href="/archives/"><i class="fa-solid fa-archive"></i><span>文章归档</span></a><a class="rightMenu-item menu-link" href="/categories/"><i class="fa-solid fa-folder-open"></i><span>文章分类</span></a><a class="rightMenu-item menu-link" href="/tags/"><i class="fa-solid fa-tags"></i><span>文章标签</span></a></div><div class="rightMenu-group rightMenu-line rightMenuNormal"><a class="rightMenu-item" id="menu-translate"><i class="fa-solid fa-earth-asia"></i><span>繁简切换</span></a><div class="rightMenu-item" id="menu-darkmode"><i class="fa-solid fa-moon"></i><span>切换模式</span></div></div></div><div id="rightmenu-mask"></div><div><script src="/oct25-xxxxx/js/utils.js"></script><script src="/oct25-xxxxx/js/main.js"></script><script src="/oct25-xxxxx/js/tw_cn.js"></script><script src="https://cdn.jsdelivr.net/npm/@fancyapps/ui/dist/fancybox.umd.js"></script><script src="/oct25-xxxxx/js/search/local-search.js"></script><div class="js-pjax"></div><script defer src="https://npm.elemecdn.com/jquery@latest/dist/jquery.min.js"></script><script defer data-pjax src="/js/rightMenu.js"></script><script defer="defer" id="ribbon" src="https://cdn.jsdelivr.net/npm/butterfly-extsrc@1/dist/canvas-ribbon.min.js" size="150" alpha="0.6" zIndex="-1" mobile="false" data-click="true"></script><script src="https://cdn.jsdelivr.net/npm/butterfly-extsrc@1/dist/activate-power-mode.min.js"></script><script>POWERMODE.colorful = true;
POWERMODE.shake = true;
POWERMODE.mobile = false;
document.body.addEventListener('input', POWERMODE);
</script><script id="click-heart" src="https://cdn.jsdelivr.net/npm/butterfly-extsrc@1/dist/click-heart.min.js" async="async" mobile="false"></script><script async data-pjax src="//busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js"></script></div></body></html>